RabbitMQ入门与核心概念

著作权归作者所有，任何形式的转载都请联系作者获得授权并注明出处。

初识RabbitMQ

rabbitMQ是一个开源的消息代理和队列服务器，用于通过普通协议在完全不同的应用之间共享数据，RabbitMQ是使用Erlang语言来编写的，并且RabbitMQ是基于AMQP协议的。

哪些大厂在用RabbitMQ，为什么？

• 滴滴、美团、头条、去哪儿、ppmoney
• 开源、性能优秀，稳定性保障
• 提供可靠的消息投递模式（confirm）、返回模式（return）
• 与SpringAMQP完美的整合，API丰富
• 集群模式丰富，表达式配置，HA模式，镜像队列模型
• 保证消息数据不丢失的前提做到高可靠性、可用性

RabbitMQ高性能的原因

• Erlang语言最初是用于交换机领域的架构模式，这样的话RabbitMQ在Broker之间进行数据交互的性能是优秀的。
• Erlang语言的特点：Erlang有着和原生Socket一样的延迟

什么是AMQP高级消息队列协议？

AMQP定义：是具有现代特征的二进制协议。是一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计，是一个规范，RabbitMQ就是这个规范的一种实现。

AMQP协议模型

AMQP核心概念

• Server:又称Broker，接受客户端的连接，实现AMQP实体服务
• Connection：连接，应用程序与Broker的网络连接
• Channel：网络信道，几乎所有的操作都在Channel中进行，Channel是进行消息读写的通道。客户端可建立多个Channel，每个Channel代表一个会话任务。
• Message：消息，服务器与应用程序之间传递的数据，由Properties和Body组成。Properties可以对消息进行修饰，比如消息的优先级，延迟等高级特性；Body则是消息体内容。
• Virtual host：虚拟地址，用于进行逻辑隔离，最上层的消息路由。一个Virtual host里面可以有若干个Exchange和Queue，同一个Virtual host里面不能有相同名称的Exchange和Queue（例如：redis有16个数据库，也是逻辑隔离的概念）
• Exchange：交换机，接受消息，根据路由键转发消息到绑定的队列
• Binding：Exchange和Queue之间的虚拟连接，binding中可以包含routing key
• Routing key：一个路由规则，虚拟机可以用它来确定如何路由一个特定消息
• Queue：也称为Message Queue，消息队列，保存消息并将它们转发给消费者

RabbitMQ的整体架构

生产者只需要关注把消息投递到指定的exchange即可，消费者监听指定的队列就可以了。通过这个图可以发现，生产者不关心我把消息投递到哪个队列，消费者也不关系我这个队列的消息是从哪个exchange来的，这两者完全没有耦合的情况。那么它们是怎么流转的，是通过exchange和queue的绑定关系。

消息流转图

RabbitMQ的安装

rabbitMQ之安装和配置(一)
rabbitmq之配置文件详解（二）

安装完成之后我们可以看到有三个与rabbitmq相关的命令

服务的启动：rabbitmq-server start & 
服务的停止：rabbitmqctl stop_app
管理插件： rabbitmq-plugins enable rabbitmq_management

验证是否启动成功，出现如下内容则为成功

使用rabbitmq-plugins list命令查看所有已经存在的插件

查看管理控制台页面

命令行与管控台的关系

基本操作

rabbitmqctl stop_app  #关闭应用
rabbitmqctl start_app  #启动应用
rabbitmqctl status   #节点状态
rabbitmqctl add_user username password  #增加用户
rabbitmqctl list_users   #列出所有用户
rabbitmqctl delete_user username  #删除用户
rabbitmqctl clear_permissions -p vhostpath username  #清除用户权限
rabbitmqctl list_user_permissions username  #列出用户权限
rabbitmqctl change_password username newpassword  #修改密码
# 设置用户权限
rabbitmqctl set_permissions -p vhostpath username ".*"".*"".*"
rabbitmqctl add_vhost vhostpath  #创建虚拟主机
rabbitmqctl list_vhosts   #列出所有虚拟主机
rabbitmqctl list_permissions -p vhostpath  #列出虚拟主机上所有权限
rabbitmqctl delete_vhost vhostpath  #删除虚拟主机
rabbitmqctl list_queues   #查看所有队列信息
rabbitmqctl -p vhostpath purge_queue blue   #清除队列里的消息

高级操作

#移除所有数据，要在rabbitmqctl stop_app 之后使用
rabbitmqctl reset 
#[--ram] 指定节点的存储模式，ram是内存基本存储
rabbitmqctl join_cluster <clusternode> [--ram]  #组成集群命令
rabbitmqctl cluster_status   #查看集群状态
# 修改集群节点的存储形式   disc磁盘 ram内存
rabbitmqctl change_cluster_node_type disc|ram 
# 忘记节点（摘除节点）
rabbitmqctl forget_cluster_node [--offline]  
#修改节点名称
rabbitmqctl rename_cluster_node oldnode1 newnode1 [oldnode2] [newnode2...]

管控台有的操作，ctl命令行也都有

实例代码操作

maven 依赖

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>3.6.5</version>
</dependency>

生产者发布消息

package com.example.rabbitmq.quickstart;


import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Procuder {

    public static void main(String[] args) throws Exception {

        // 1 创建一个ConnectionFactory,并进行配置
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");

        //2 通过连接工程创建连接
        Connection connection = connectionFactory.newConnection();

        // 3 通过connection创建一个channel
        Channel channel = connection.createChannel();

        for (int i = 0; i < 5; i++) {
            String msg = "hello world!";
            // 1 exchange   2 routing key
            channel.basicPublish("", "test01", null, msg.getBytes());
        }

        channel.close();
        connection.close();
    }
}

消费者订阅处理消息

package com.example.rabbitmq.quickstart;

import com.rabbitmq.client.*;
import com.rabbitmq.client.QueueingConsumer.Delivery;

public class Consumer {
    public static void main(String[] args) throws Exception {
        // 1 创建一个ConnectionFactory,并进行配置
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");

        //2 通过连接工程创建连接
        Connection connection = connectionFactory.newConnection();

        // 3 通过connection创建一个channel
        Channel channel = connection.createChannel();
        //4 声明（创建）队列
        String queueName = "test01";
        channel.queueDeclare(queueName, true, false, false, null);
        //5 创建消费者
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
        // 6 设置channel
        channel.basicConsume(queueName, true, queueingConsumer);

        while (true) {
            Delivery delivery = queueingConsumer.nextDelivery();
            String msg = new String(delivery.getBody());
            System.out.println("消费端消息：" + msg);
            Envelope envelope = delivery.getEnvelope();
        }
    }
}

exchange交换机

交换机属性

• Name: 交换机名称，
• Type: 交换机类型direct、topic、fanout、headers
• Durability: 是否需要持久化，true为持久化
• Auto Delete：当最后一个绑定到Exchange上的队列删除后，自动删除该Exchange
• Internal：当前Exchange是否用于RabbitMQ内部使用，默认为false
• ArgArgumens：扩展参数，用于扩展AMQP协议自制定化使用

Direct Exchange

所有发送到Direct Exchange的消息被转发到RouteKey中指定到Queue

⚠️注意：Direct模式可以使用RabbitMQ自带的Exchange：default Exchange，所以不需要将Exchange进行任何绑定（binding）操作，消息传递时，RouteKey必须完全匹配才会被队列接收，否则该消息会被抛弃。

示例代码

package com.bfxy.rabbitmq.api.exchange.direct;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;

public class ConsumerDirectExchange {
	public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory() ;  
        
        connectionFactory.setHost("192.168.11.76");
        connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
        connectionFactory.setAutomaticRecoveryEnabled(true);
        connectionFactory.setNetworkRecoveryInterval(3000);
        Connection connection = connectionFactory.newConnection();
        
        Channel channel = connection.createChannel();  
		//4 声明
		String exchangeName = "test_direct_exchange";
		String exchangeType = "direct";
		String queueName = "test_direct_queue";
		String routingKey = "test.direct";
		
		//表示声明了一个交换机
		channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
		//表示声明了一个队列
		channel.queueDeclare(queueName, false, false, false, null);
		//建立一个绑定关系:
		channel.queueBind(queueName, exchangeName, routingKey);
		
        //durable 是否持久化消息
        QueueingConsumer consumer = new QueueingConsumer(channel);
        //参数：队列名称、是否自动ACK、Consumer
        channel.basicConsume(queueName, true, consumer);  
        //循环获取消息  
        while(true){  
            //获取消息，如果没有消息，这一步将会一直阻塞  
            Delivery delivery = consumer.nextDelivery();  
            String msg = new String(delivery.getBody());    
            System.out.println("收到消息：" + msg);  
        } 
	}
}

package com.bfxy.rabbitmq.api.exchange.direct;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class ProducerDirectExchange {
	public static void main(String[] args) throws Exception {
		//1 创建ConnectionFactory
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("192.168.11.76");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
		//2 创建Connection
		Connection connection = connectionFactory.newConnection();
		//3 创建Channel
		Channel channel = connection.createChannel();  
		//4 声明
		String exchangeName = "test_direct_exchange";
		String routingKey = "test.direct111";
		//5 发送
		
		String msg = "Hello World RabbitMQ 4  Direct Exchange Message 111 ... ";
		channel.basicPublish(exchangeName, routingKey , null , msg.getBytes()); 		
	}
}

Topic Exchange

所有发送到Topic Exchange的消息会被转发到所有关心该RouteKey中指定Topic的Queue上

Exchange将RouteKey和某Topic进行模糊匹配，此时队列需要绑定一个Topic

⚠️注意：可以使用通配符进行模糊匹配

符号 "#" 匹配一个或多个词   
符号 "*" 匹配不多不少一个词
例如："log.#" 能够匹配到 "log.info.oa"
      "log.*" 只会匹配到 "log.error"

示例代码

package com.bfxy.rabbitmq.api.exchange.topic;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;

public class ConsumerTopicExchange {
	public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory() ;  
        
        connectionFactory.setHost("192.168.11.76");
        connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
        connectionFactory.setAutomaticRecoveryEnabled(true);
        connectionFactory.setNetworkRecoveryInterval(3000);
        Connection connection = connectionFactory.newConnection();
        
        Channel channel = connection.createChannel();  
		//4 声明
		String exchangeName = "test_topic_exchange";
		String exchangeType = "topic";
		String queueName = "test_topic_queue";
		//String routingKey = "user.*";
		String routingKey = "user.*";
		// 1 声明交换机 
		channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
		// 2 声明队列
		channel.queueDeclare(queueName, false, false, false, null);
		// 3 建立交换机和队列的绑定关系:
		channel.queueBind(queueName, exchangeName, routingKey);
		
        //durable 是否持久化消息
        QueueingConsumer consumer = new QueueingConsumer(channel);
        //参数：队列名称、是否自动ACK、Consumer
        channel.basicConsume(queueName, true, consumer);  
        //循环获取消息  
        while(true){  
            //获取消息，如果没有消息，这一步将会一直阻塞  
            Delivery delivery = consumer.nextDelivery();  
            String msg = new String(delivery.getBody());    
            System.out.println("收到消息：" + msg);  
        } 
	}
}

package com.bfxy.rabbitmq.api.exchange.topic;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class ProducerTopicExchange {
	public static void main(String[] args) throws Exception {
		
		//1 创建ConnectionFactory
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("192.168.11.76");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
		//2 创建Connection
		Connection connection = connectionFactory.newConnection();
		//3 创建Channel
		Channel channel = connection.createChannel();  
		//4 声明
		String exchangeName = "test_topic_exchange";
		String routingKey1 = "user.save";
		String routingKey2 = "user.update";
		String routingKey3 = "user.delete.abc";
		//5 发送
		String msg = "Hello World RabbitMQ 4 Topic Exchange Message ...";
		channel.basicPublish(exchangeName, routingKey1 , null , msg.getBytes()); 
		channel.basicPublish(exchangeName, routingKey2 , null , msg.getBytes()); 	
		channel.basicPublish(exchangeName, routingKey3 , null , msg.getBytes()); 
		channel.close();  
        connection.close();  
	}
}

Fanout Exchange

• 不处理路由键，只需要简单到将队列绑定到交换机上
• 发送到交换机的消息都会被转发到与该交换机绑定到所有队列上
• Fanout交换机转发消息上最快的

示例代码

package com.bfxy.rabbitmq.api.exchange.fanout;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;

public class ConsumerFanoutExchange {
	public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory() ;  
        
        connectionFactory.setHost("192.168.11.76");
        connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
        connectionFactory.setAutomaticRecoveryEnabled(true);
        connectionFactory.setNetworkRecoveryInterval(3000);
        Connection connection = connectionFactory.newConnection();
        
        Channel channel = connection.createChannel();  
		//4 声明
		String exchangeName = "test_fanout_exchange";
		String exchangeType = "fanout";
		String queueName = "test_fanout_queue";
		String routingKey = "";	//不设置路由键
		channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);
		channel.queueDeclare(queueName, false, false, false, null);
		channel.queueBind(queueName, exchangeName, routingKey);
		
        //durable 是否持久化消息
        QueueingConsumer consumer = new QueueingConsumer(channel);
        //参数：队列名称、是否自动ACK、Consumer
        channel.basicConsume(queueName, true, consumer); 
        //循环获取消息  
        while(true){  
            //获取消息，如果没有消息，这一步将会一直阻塞  
            Delivery delivery = consumer.nextDelivery();  
            String msg = new String(delivery.getBody());    
            System.out.println("收到消息：" + msg);  
        } 
	}
}

package com.bfxy.rabbitmq.api.exchange.fanout;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class ProducerFanoutExchange {
	public static void main(String[] args) throws Exception {
		//1 创建ConnectionFactory
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("192.168.11.76");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");
		
		//2 创建Connection
		Connection connection = connectionFactory.newConnection();
		//3 创建Channel
		Channel channel = connection.createChannel();  
		//4 声明
		String exchangeName = "test_fanout_exchange";
		//5 发送
		for(int i = 0; i < 10; i ++) {
			String msg = "Hello World RabbitMQ 4 FANOUT Exchange Message ...";
			channel.basicPublish(exchangeName, "", null , msg.getBytes()); 			
		}
		channel.close();  
        connection.close();  
	}
}

基础概念讲解

Binding-绑定

• Exchange和Exchange、Queue之间的连接关系
• Binding可以包含RoutingKey或者参数

Queue-消息队列

• 消息队列，实际存储消息数据
• Durability：是否持久化，Durable：是，Transient：否
• Auto Delete：如选yes，代表当最后一个监听被移除之后，该Queue会自动被删除

Message-消息

• 服务器与应用程序之间传递当数据
• 本质上就是一段数据，由propertie和Payload（Body）组成
• 常用属性：delivery mode、headers（自定义属性）、content_type、content_enconding、priority、correlation_id、reply_to、expiration、message_id、timestamp、type、user_id、app_id、cluster_id

Virtual host-虚拟主机

• 虚拟地址，用于进行逻辑隔离，最上层当消息路由
• 一个Virtual host里面可以由若干个Exchange和Queue
• 同一个Virtual host里面不能有相同名称的Exchange或Queue

更多内容阅读：
rabbitmq核心概念总结
中文文档