2.1初识RabbitMQ

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

为什么选用RabbitMQ

开源、性能优秀、稳定性保障提供可靠性消息投递模式（confirm）、返回模式（return）SpringAMQP完美的整合、API丰富集群模式丰富、表达式配置、HA模式（高可用）、镜像队列模式保证数据不丢失的前提下保证高可靠性和可用性

RabbitMQ高性能的原因

Erlang语言最初在于交换机领域的架构模式，这样使得RabbitMQ在Broker之间进行数据交互的性能非常优秀Erlang有着和原生Socket一样的延迟

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

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

AMQP协议模型 - Publisher application生产者应用服务

Consumer application消费者应用服务,只需要监控消息队列，从队列中取数据消息投递过程为Server–>Virtual host–>Exchange

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或QueueExchange：交换机、接收消息。根据路由键转发消息到绑定的队列Binding：Exchange和Queue之间的虚拟连接，binding中可以包含routing keyRouting key：一个路由规则，虚拟机可用它确定如何路由一个特定消息Queue：也称Message Queue，消息队列，保存消息并将它们转发给消费者

RabbitMQ的整体架构

2.2RabbitMQ的配置

配置文件

vim ebin/rabbit.app

修改如下： 相关命令

服务的启动：rabbitmq-server start &（&表明后台启动）服务的停止 rabbitmqctl stop_app管理插件：rabbitmq-plugins enable rabbitmq_management访问地址：http://192.168.11.76:15672/

验证是否启动成功

输入：lsof -i：端口 表明查询谁在使用某端口 lsof -i:5672 输出： COMMAND PID USER FD TYPE DEVICE SIZE/OFF NODE NAME beam.smp 7503 root 56u IPv6 44087 0t0 TCP *:amqp (LISTEN)

查看插件

rabbitmq-plugins list

使用插件rabbitmq_management

rabbitmq-plugins enable rabbitmq_management 访问地址： http://192.168.11.76:15672/ //注意开放15672的端口号 //查看防火墙所有开放的端口 firewall-cmd --zone=public --list-ports //开放端口 类似如下 firewall-cmd --zone=public --add-port=5672/tcp --permanent # 开放5672端口 firewall-cmd --zone=public --remove-port=5672/tcp --permanent #关闭5672端口 firewall-cmd --reload # 配置立即生效

访问页面显示 username：guest password：guest 这是在前面配置文件中改过的

2.3命令行与管控台-基础操作

开启和关闭

关闭应用：rabbitmqctl stop_app启动应用：rabbitmqctl start_app节点状态：rabbitmqctl status

用户操作

添加用户：rabbitmqctl add_user username password列出所有用户：rabbitmqctl list_users删除用户：rabbitmqctl delete_user username清除用户权限：rabbitmq 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

队列操作

查看所有队列消息：rabbitctl list_queues清除队列里的消息：rabbitmqctl -p vhostpath purge_queue blue

2.4命令行与管控台-高级操作

移除所有数据：rabbitmqctl reset，注意在rabbitmqctl stop_app之后使用组成集群操作：rabbitmqctl join_cluster < clusternode > [–ram]，ram表明存在内存中查看集群状态：rabbitmqctl cluster_status修改集群节点的存储形式：rabbitmqctl change_cluster_node_type disc | ram摘除节点：rabbitmqctl forget_cluster_node [–offline]，用于忘记故障节点修改节点名称：rabbitmqctl rename_cluster_node oldnode1 newnode1 …

2.5急速入门-消息生产与消费

ConnectionFactory：获取连接工厂Connection：一个连接Channel：数据通信信道，可发送和接受消息Queue：具体消息队列Producer&Consumer 生产者和消费者

消费者代码

public class Producer { public static void main(String[] args) throws Exception { //创建一个ConnectionFactory ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("192.168.174.10"); connectionFactory.setPort(5672); connectionFactory.setVirtualHost("/"); //通过连接工厂创建连接 Connection connection = connectionFactory.newConnection(); //通过connection创建一个Channel Channel channel = connection.createChannel(); //通过channel发送数据 前两个参数时 for(int i=0; i < 5; i++){ String msg = "Hello RabbitMQ!"; //1 exchange 2 routingKey channel.basicPublish("", "test001", null, msg.getBytes()); } //记得关闭连接 channel.close(); connection.close(); }

生产者代码

public class Consumer { public static void main(String[] args) throws Exception { //创建一个ConnectionFactory ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("192.168.174.10"); connectionFactory.setPort(5672); connectionFactory.setVirtualHost("/"); //通过连接工厂创建连接 Connection connection = connectionFactory.newConnection(); //通过connection创建一个Channel Channel channel = connection.createChannel(); //声明一个队列 第三个参数配置是否独占 保证消费顺序性， String queueName = "test001"; channel.queueDeclare(queueName, true, false, false, null); //5 创建消费者 QueueingConsumer queueingConsumer = new QueueingConsumer(channel); //6 设置Channel channel.basicConsume(queueName, true, queueingConsumer); while(true){ //7 获取消息 QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery(); String msg = new String(delivery.getBody()); System.err.println("消费端: " + msg); //Envelope envelope = delivery.getEnvelope(); } } }

注意：

2.6Exchange交换机

Exchange：接受消息，并根据路由键转发消息所绑定的队列

蓝框：生产者绿框：消费者黄框：路由键

交换机属性

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

Direct Exchange

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

生产者代码：

public class Producer4DirectExchange { 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.direct"; //5 发送 String msg = "Hello World RabbitMQ 4 Direct Exchange Message 111 ... "; channel.basicPublish(exchangeName, routingKey , null , msg.getBytes()); } }

消费者代码

public class Consumer4DirectExchange { 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); } } }

Topic Exchange

所有发送到Topic Exchange的消息被转发到所有关心RouteKey中指定Topic的Queue上Exchange将RouteKey和某Topic进行模糊匹配，此时队列需要绑定一个Topic可以使用通配符进行模糊匹配，# 匹配一个或多个词，* 匹配不多不少一个词

生产者代码

public class Producer4TopicExchange { 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(); } }

消费者代码

public class Consumer4TopicExchange { 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); } } }

Fanout Exchange

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

生产者代码

public class Producer4FanoutExchange { 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(); } }

消费者代码

public class Consumer4FanoutExchange { 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); } } }

2.7其他概念

Binding-绑定

Exchange和Exchange、Queue之间的连接关系Bind包含RoutingKey或者参数

Queue-消息队列

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

Message-消息

服务器和应用程序之间传送的数据本质上就是一段数据，由Properties和Payload（Body）组成常用属性：delivery mode（可以选择是否进行持久化）、header（自定义属性）其他属性： content_type、content_encoding、prioritycorrelation_id 可以把它当成消息的idreply_to 消息失败了返回哪一个队列expiration 消息过期时间message_id 消息的idtimestamp、type、user_id、app_id、cluster_id

message相关代码示例

主要是给消息设置了属性

生产者代码

public class Procuder { 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 = connectionFactory.newConnection(); //3 通过connection创建一个Channel Channel channel = connection.createChannel(); Map<String, Object> headers = new HashMap<>(); headers.put("my1", "111"); headers.put("my2", "222"); AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder() .deliveryMode(2) .contentEncoding("UTF-8") .expiration("10000") .headers(headers) .build(); //4 通过Channel发送数据 for(int i=0; i < 5; i++){ String msg = "Hello RabbitMQ!"; //1 exchange 2 routingKey channel.basicPublish("", "test001", properties, msg.getBytes()); } //5 记得要关闭相关的连接 channel.close(); connection.close(); } }

消费者代码

public class Consumer { 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 = connectionFactory.newConnection(); //3 通过connection创建一个Channel Channel channel = connection.createChannel(); //4 声明（创建）一个队列 String queueName = "test001"; channel.queueDeclare(queueName, true, false, false, null); //5 创建消费者 QueueingConsumer queueingConsumer = new QueueingConsumer(channel); //6 设置Channel channel.basicConsume(queueName, true, queueingConsumer); while(true){ //7 获取消息 Delivery delivery = queueingConsumer.nextDelivery(); String msg = new String(delivery.getBody()); System.err.println("消费端: " + msg); Map<String, Object> headers = delivery.getProperties().getHeaders(); System.err.println("headers get my1 value: " + headers.get("my1")); //Envelope envelope = delivery.getEnvelope(); } } }

Virtual host-虚拟主机

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

2.8本章小结

牛逼就对了