MQ:全称为Message Queue，消息队列是应用程序和应用程序之间的通信方法.
为什么使用MQ? 在项目中，可将一些无需即时返回且耗时的操作提取出来，进行异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量。 应用场景 任务异步处理应用程序解耦合削峰填谷MQ是消息通信的模型；实现MQ的大致有两种主流方式：AMQP、JMS
AMQP是一种协议，更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。
JMS即Java消息服务（JavaMessage Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。
RabbitMQ是由erlang语言开发，基于AMQP（Advanced Message Queue 高级消息队列协议）协议实现的消息队列，它是一种应用程序之间的通信方法，消息队列在分布式系统开发中应用非常广泛. RabbitMQ官网
RabbitMQ提供了6种模式：
简单模式work模式Publish/Subscribe发布与订阅模式Routing路由模式Topics主题模式RPC远程调用模式(属于远程调用)官网介绍
生产者1: 1 pom.xml中导入坐标
<dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.6.0</version> </dependency>2 编写消息生产者
package com.cf.rabbitmq.simple; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class Producer { static final String QUEUE_NAME = "simple_queue"; public static void main(String[] args) throws Exception { //创建连接工厂 ConnectionFactory connectionFactory = new ConnectionFactory(); //主机地址;默认为 localhost connectionFactory.setHost("localhost"); //连接端口;默认为 5672 connectionFactory.setPort(5672); //虚拟主机名称;默认为 / connectionFactory.setVirtualHost("/itcast"); //连接用户名；默认为guest connectionFactory.setUsername("cf"); //连接密码；默认为guest connectionFactory.setPassword("12345678"); //创建连接 Connection connection = connectionFactory.newConnection(); // 创建频道 Channel channel = connection.createChannel(); // 声明（创建）队列 /** * 参数1：队列名称 * 参数2：是否定义持久化队列 * 参数3：是否独占本次连接 * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */ channel.queueDeclare(QUEUE_NAME, true, false, false, null); // 要发送的信息 String message = "你好啊；小白兔！"; /** * 参数1：交换机名称，如果没有指定则使用默认Default Exchage * 参数2：路由key,简单模式可以传递队列名称 * 参数3：消息其它属性 * 参数4：消息内容 */ channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println("已发送消息：" + message); // 关闭资源 channel.close(); connection.close(); } }消费者1: 1 抽取创建connection的工具类
package com.cf.rabbitmq.util; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class ConnectionUtil { public static Connection getConnection() throws Exception { //创建连接工厂 ConnectionFactory connectionFactory = new ConnectionFactory(); //主机地址;默认为 localhost connectionFactory.setHost("localhost"); //连接端口;默认为 5672 connectionFactory.setPort(5672); //虚拟主机名称;默认为 / connectionFactory.setVirtualHost("/cf"); //连接用户名；默认为guest connectionFactory.setUsername("cf"); //连接密码；默认为guest connectionFactory.setPassword("12345678"); //创建连接 return connectionFactory.newConnection(); } }2 编写消息的消费者
package com.cf.rabbitmq.simple; import com.cf.rabbitmq.util.ConnectionUtil; import com.rabbitmq.client.*; import java.io.IOException; public class Consumer { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); // 声明（创建）队列 /** * 参数1：队列名称 * 参数2：是否定义持久化队列 * 参数3：是否独占本次连接 * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */ channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null); //创建消费者；并设置消息处理 DefaultConsumer consumer = new DefaultConsumer(channel){ @Override /** * consumerTag 消息者标签，在channel.basicConsume时候可以指定 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) * properties 属性信息 * body 消息 */ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { //路由key System.out.println("路由key为：" + envelope.getRoutingKey()); //交换机 System.out.println("交换机为：" + envelope.getExchange()); //消息id System.out.println("消息id为：" + envelope.getDeliveryTag()); //收到的消息 System.out.println("接收到的消息为：" + new String(body, "utf-8")); } }; //监听消息 /** * 参数1：队列名称 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 * 参数3：消息接收到后回调 */ channel.basicConsume(Producer.QUEUE_NAME, true, consumer); //不关闭资源，应该一直监听消息 //channel.close(); //connection.close(); } }总结:
入门案例中使用的是简单模式. P：生产者，也就是要发送消息的程序C：消费者：消息的接受者，会一直等待消息到来queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息应用场景: 对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度.
Work Queues与入门程序的简单模式的代码是几乎一样的；可以完全复制，并复制多一个消费者进行多个消费者同时消费消息的测试. 消费者2:
package com.cf.rabbitmq.work; import com.cf.rabbitmq.util.ConnectionUtil; import com.rabbitmq.client.*; import java.io.IOException; public class Consumer2 { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); // 声明（创建）队列 /** * 参数1：队列名称 * 参数2：是否定义持久化队列 * 参数3：是否独占本次连接 * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */ channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null); //一次只能接收并处理一个消息 //为了让两个消费者都有接受消息才设置,可选 channel.basicQos(1); //创建消费者；并设置消息处理 DefaultConsumer consumer = new DefaultConsumer(channel){ @Override /** * consumerTag 消息者标签，在channel.basicConsume时候可以指定 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) * properties 属性信息 * body 消息 */ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { try { //路由key System.out.println("路由key为：" + envelope.getRoutingKey()); //交换机 System.out.println("交换机为：" + envelope.getExchange()); //消息id System.out.println("消息id为：" + envelope.getDeliveryTag()); //收到的消息 System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8")); Thread.sleep(1000); //确认消息 channel.basicAck(envelope.getDeliveryTag(), false); } catch (InterruptedException e) { e.printStackTrace(); } } }; //监听消息 /** * 参数1：队列名称 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 * 参数3：消息接收到后回调 */ channel.basicConsume(Producer.QUEUE_NAME, false, consumer); } }总结:
在一个队列中如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系总结: Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失.
生产者:
package com.cf.rabbitmq.ps; import com.cf.rabbitmq.util.ConnectionUtil; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; /** * 发布与订阅使用的交换机类型为：fanout */ public class Producer { //交换机名称 static final String FANOUT_EXCHAGE = "fanout_exchange"; //队列名称 static final String FANOUT_QUEUE_1 = "fanout_queue_1"; //队列名称 static final String FANOUT_QUEUE_2 = "fanout_queue_2"; public static void main(String[] args) throws Exception { //创建连接 Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); /** * 声明交换机 * 参数1：交换机名称 * 参数2：交换机类型，fanout、topic、direct、headers */ channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT); // 声明（创建）队列 /** * 参数1：队列名称 * 参数2：是否定义持久化队列 * 参数3：是否独占本次连接 * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */ channel.queueDeclare(FANOUT_QUEUE_1, true, false, false, null); channel.queueDeclare(FANOUT_QUEUE_2, true, false, false, null); //队列绑定交换机 channel.queueBind(FANOUT_QUEUE_1, FANOUT_EXCHAGE, ""); channel.queueBind(FANOUT_QUEUE_2, FANOUT_EXCHAGE, ""); for (int i = 1; i <= 10; i++) { // 发送信息 String message = "你好；小兔子！发布订阅模式--" + i; /** * 参数1：交换机名称，如果没有指定则使用默认Default Exchage * 参数2：路由key,简单模式可以传递队列名称 * 参数3：消息其它属性 * 参数4：消息内容 */ channel.basicPublish(FANOUT_EXCHAGE, "", null, message.getBytes()); System.out.println("已发送消息：" + message); } // 关闭资源 channel.close(); connection.close(); } }消费者1:
package com.cf.rabbitmq.ps; import com.cf.rabbitmq.util.ConnectionUtil; import com.rabbitmq.client.*; import java.io.IOException; public class Consumer1 { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); //声明交换机 channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT); // 声明（创建）队列 /** * 参数1：队列名称 * 参数2：是否定义持久化队列 * 参数3：是否独占本次连接 * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */ channel.queueDeclare(Producer.FANOUT_QUEUE_1, true, false, false, null); //队列绑定交换机 channel.queueBind(Producer.FANOUT_QUEUE_1, Producer.FANOUT_EXCHAGE, ""); //创建消费者；并设置消息处理 DefaultConsumer consumer = new DefaultConsumer(channel){ @Override /** * consumerTag 消息者标签，在channel.basicConsume时候可以指定 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) * properties 属性信息 * body 消息 */ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { //路由key System.out.println("路由key为：" + envelope.getRoutingKey()); //交换机 System.out.println("交换机为：" + envelope.getExchange()); //消息id System.out.println("消息id为：" + envelope.getDeliveryTag()); //收到的消息 System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8")); } }; //监听消息 /** * 参数1：队列名称 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 * 参数3：消息接收到后回调 */ channel.basicConsume(Producer.FANOUT_QUEUE_1, true, consumer); } }消费者2:
package com.cf.rabbitmq.ps; import com.cf.rabbitmq.util.ConnectionUtil; import com.rabbitmq.client.*; import java.io.IOException; public class Consumer2 { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); //声明交换机 channel.exchangeDeclare(Producer.FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT); // 声明（创建）队列 /** * 参数1：队列名称 * 参数2：是否定义持久化队列 * 参数3：是否独占本次连接 * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */ channel.queueDeclare(Producer.FANOUT_QUEUE_2, true, false, false, null); //队列绑定交换机 channel.queueBind(Producer.FANOUT_QUEUE_2, Producer.FANOUT_EXCHAGE, ""); //创建消费者；并设置消息处理 DefaultConsumer consumer = new DefaultConsumer(channel){ @Override /** * consumerTag 消息者标签，在channel.basicConsume时候可以指定 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) * properties 属性信息 * body 消息 */ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { //路由key System.out.println("路由key为：" + envelope.getRoutingKey()); //交换机 System.out.println("交换机为：" + envelope.getExchange()); //消息id System.out.println("消息id为：" + envelope.getDeliveryTag()); //收到的消息 System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8")); } }; //监听消息 /** * 参数1：队列名称 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 * 参数3：消息接收到后回调 */ channel.basicConsume(Producer.FANOUT_QUEUE_2, true, consumer); } }总结: 交换机需要与队列进行绑定，绑定之后；一个消息可以被多个消费者都收到.
发布订阅模式与工作队列模式的区别:
工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息(底层使用默认交换机)发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑 定到默认的交换机特点:
队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）消息的发送方在 向 Exchange发送消息时，也必须指定消息的 RoutingKeyExchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息 P：生产者，向Exchange发送消息，发送消息时，会指定一个routing keyX：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列C1：消费者，其所在队列指定了需要routing key 为 error 的消息C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息在编码上与 Publish/Subscribe发布与订阅模式 的区别是交换机的类型为：Direct，还有队列绑定交换机的时候需要指定routing key
生产者:
package com.cf.rabbitmq.routing; import com.cf.rabbitmq.util.ConnectionUtil; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; /** * 路由模式的交换机类型为：direct */ public class Producer { //交换机名称 static final String DIRECT_EXCHAGE = "direct_exchange"; //队列名称 static final String DIRECT_QUEUE_INSERT = "direct_queue_insert"; //队列名称 static final String DIRECT_QUEUE_UPDATE = "direct_queue_update"; public static void main(String[] args) throws Exception { //创建连接 Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); /** * 声明交换机 * 参数1：交换机名称 * 参数2：交换机类型，fanout、topic、direct、headers */ channel.exchangeDeclare(DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT); // 声明（创建）队列 /** * 参数1：队列名称 * 参数2：是否定义持久化队列 * 参数3：是否独占本次连接 * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */ channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null); channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null); //队列绑定交换机 channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHAGE, "insert"); channel.queueBind(DIRECT_QUEUE_UPDATE, DIRECT_EXCHAGE, "update"); // 发送信息 String message = "新增了商品。路由模式；routing key 为 insert " ; /** * 参数1：交换机名称，如果没有指定则使用默认Default Exchage * 参数2：路由key,简单模式可以传递队列名称 * 参数3：消息其它属性 * 参数4：消息内容 */ channel.basicPublish(DIRECT_EXCHAGE, "insert", null, message.getBytes()); System.out.println("已发送消息：" + message); // 发送信息 message = "修改了商品。路由模式；routing key 为 update" ; /** * 参数1：交换机名称，如果没有指定则使用默认Default Exchage * 参数2：路由key,简单模式可以传递队列名称 * 参数3：消息其它属性 * 参数4：消息内容 */ channel.basicPublish(DIRECT_EXCHAGE, "update", null, message.getBytes()); System.out.println("已发送消息：" + message); // 关闭资源 channel.close(); connection.close(); } }消费者1:
package com.cf.rabbitmq.routing; import com.cf.rabbitmq.util.ConnectionUtil; import com.rabbitmq.client.*; import java.io.IOException; public class Consumer1 { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); //声明交换机 channel.exchangeDeclare(Producer.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT); // 声明（创建）队列 /** * 参数1：队列名称 * 参数2：是否定义持久化队列 * 参数3：是否独占本次连接 * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */ channel.queueDeclare(Producer.DIRECT_QUEUE_INSERT, true, false, false, null); //队列绑定交换机 channel.queueBind(Producer.DIRECT_QUEUE_INSERT, Producer.DIRECT_EXCHAGE, "insert"); //创建消费者；并设置消息处理 DefaultConsumer consumer = new DefaultConsumer(channel){ @Override /** * consumerTag 消息者标签，在channel.basicConsume时候可以指定 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) * properties 属性信息 * body 消息 */ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { //路由key System.out.println("路由key为：" + envelope.getRoutingKey()); //交换机 System.out.println("交换机为：" + envelope.getExchange()); //消息id System.out.println("消息id为：" + envelope.getDeliveryTag()); //收到的消息 System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8")); } }; //监听消息 /** * 参数1：队列名称 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 * 参数3：消息接收到后回调 */ channel.basicConsume(Producer.DIRECT_QUEUE_INSERT, true, consumer); } }消费者2:
package com.cf.rabbitmq.routing; import com.cf.rabbitmq.util.ConnectionUtil; import com.rabbitmq.client.*; import java.io.IOException; public class Consumer2 { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); //声明交换机 channel.exchangeDeclare(Producer.DIRECT_EXCHAGE, BuiltinExchangeType.DIRECT); // 声明（创建）队列 /** * 参数1：队列名称 * 参数2：是否定义持久化队列 * 参数3：是否独占本次连接 * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */ channel.queueDeclare(Producer.DIRECT_QUEUE_UPDATE, true, false, false, null); //队列绑定交换机 channel.queueBind(Producer.DIRECT_QUEUE_UPDATE, Producer.DIRECT_EXCHAGE, "update"); //创建消费者；并设置消息处理 DefaultConsumer consumer = new DefaultConsumer(channel){ @Override /** * consumerTag 消息者标签，在channel.basicConsume时候可以指定 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) * properties 属性信息 * body 消息 */ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { //路由key System.out.println("路由key为：" + envelope.getRoutingKey()); //交换机 System.out.println("交换机为：" + envelope.getExchange()); //消息id System.out.println("消息id为：" + envelope.getDeliveryTag()); //收到的消息 System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8")); } }; //监听消息 /** * 参数1：队列名称 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 * 参数3：消息接收到后回调 */ channel.basicConsume(Producer.DIRECT_QUEUE_UPDATE, true, consumer); } }总结:
Routing模式要求队列在绑定交换机时要指定routing key，消息会转发到符合routing key的队列Topic类型与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！
通配符规则：
#：匹配一个或多个词*：匹配不多不少恰好1个词 e.g:item.#：能够匹配item.insert.abc 或者 item.insertitem.*：只能匹配item.insert 红色Queue：绑定的是usa.# ，因此凡是以 usa.开头的routing key 都会被匹配到黄色Queue：绑定的是#.news ，因此凡是以 .news结尾的 routing key 都会被匹配生产者:
package com.cf.rabbitmq.topic; import com.cf.rabbitmq.util.ConnectionUtil; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; /** * 通配符Topic的交换机类型为：topic */ public class Producer { //交换机名称 static final String TOPIC_EXCHAGE = "topic_exchange"; //队列名称 static final String TOPIC_QUEUE_1 = "topic_queue_1"; //队列名称 static final String TOPIC_QUEUE_2 = "topic_queue_2"; public static void main(String[] args) throws Exception { //创建连接 Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); /** * 声明交换机 * 参数1：交换机名称 * 参数2：交换机类型，fanout、topic、topic、headers */ channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC); // 发送信息 String message = "新增了商品。Topic模式；routing key 为 item.insert " ; channel.basicPublish(TOPIC_EXCHAGE, "item.insert", null, message.getBytes()); System.out.println("已发送消息：" + message); // 发送信息 message = "修改了商品。Topic模式；routing key 为 item.update" ; channel.basicPublish(TOPIC_EXCHAGE, "item.update", null, message.getBytes()); System.out.println("已发送消息：" + message); // 发送信息 message = "删除了商品。Topic模式；routing key 为 item.delete" ; channel.basicPublish(TOPIC_EXCHAGE, "item.delete", null, message.getBytes()); System.out.println("已发送消息：" + message); // 关闭资源 channel.close(); connection.close(); } }消费者1:
package com.cf.rabbitmq.topic; import com.cf.rabbitmq.util.ConnectionUtil; import com.rabbitmq.client.*; import java.io.IOException; public class Consumer1 { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); //声明交换机 channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC); // 声明（创建）队列 /** * 参数1：队列名称 * 参数2：是否定义持久化队列 * 参数3：是否独占本次连接 * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */ channel.queueDeclare(Producer.TOPIC_QUEUE_1, true, false, false, null); //队列绑定交换机 channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.update"); channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.delete"); //创建消费者；并设置消息处理 DefaultConsumer consumer = new DefaultConsumer(channel){ @Override /** * consumerTag 消息者标签，在channel.basicConsume时候可以指定 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) * properties 属性信息 * body 消息 */ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { //路由key System.out.println("路由key为：" + envelope.getRoutingKey()); //交换机 System.out.println("交换机为：" + envelope.getExchange()); //消息id System.out.println("消息id为：" + envelope.getDeliveryTag()); //收到的消息 System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8")); } }; //监听消息 /** * 参数1：队列名称 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 * 参数3：消息接收到后回调 */ channel.basicConsume(Producer.TOPIC_QUEUE_1, true, consumer); } }消费者2:
package com.cf.rabbitmq.topic; import com.cf.rabbitmq.util.ConnectionUtil; import com.rabbitmq.client.*; import java.io.IOException; public class Consumer2 { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); // 创建频道 Channel channel = connection.createChannel(); //声明交换机 channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC); // 声明（创建）队列 /** * 参数1：队列名称 * 参数2：是否定义持久化队列 * 参数3：是否独占本次连接 * 参数4：是否在不使用的时候自动删除队列 * 参数5：队列其它参数 */ channel.queueDeclare(Producer.TOPIC_QUEUE_2, true, false, false, null); //队列绑定交换机 channel.queueBind(Producer.TOPIC_QUEUE_2, Producer.TOPIC_EXCHAGE, "item.*"); //创建消费者；并设置消息处理 DefaultConsumer consumer = new DefaultConsumer(channel){ @Override /** * consumerTag 消息者标签，在channel.basicConsume时候可以指定 * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) * properties 属性信息 * body 消息 */ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { //路由key System.out.println("路由key为：" + envelope.getRoutingKey()); //交换机 System.out.println("交换机为：" + envelope.getExchange()); //消息id System.out.println("消息id为：" + envelope.getDeliveryTag()); //收到的消息 System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8")); } }; //监听消息 /** * 参数1：队列名称 * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认 * 参数3：消息接收到后回调 */ channel.basicConsume(Producer.TOPIC_QUEUE_2, true, consumer); } }总结: Topic主题模式可以实现 Publish/Subscribe发布与订阅模式 和 Routing路由模式 的功能；只是Topic在配置routing key 的时候可以使用通配符，显得更加灵活.
一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）
一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）
需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列
需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列
需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列
1 在pom.xml文件中添加依赖
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.cf</groupId> <artifactId>spring-rabbitmq-producer</artifactId> <version>1.0-SNAPSHOT</version> <dependencies> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-context</artifactId> <version>5.1.7.RELEASE</version> </dependency> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit</artifactId> <version>2.1.8.RELEASE</version> </dependency> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.12</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-test</artifactId> <version>5.1.7.RELEASE</version> </dependency> </dependencies> </project>2.1 创建rabbitmq.properties配置文件
rabbitmq.host=192.168.174.90 rabbitmq.port=5672 rabbitmq.username=cf rabbitmq.password=12345678 rabbitmq.virtual-host=/cf2.2 创建spring-rabbitmq.xml整合配置文件
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xmlns:rabbit="http://www.springframework.org/schema/rabbit" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context https://www.springframework.org/schema/context/spring-context.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd"> <!--加载配置文件--> <context:property-placeholder location="classpath:rabbitmq.properties"/> <!-- 定义rabbitmq connectionFactory --> <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}" port="${rabbitmq.port}" username="${rabbitmq.username}" password="${rabbitmq.password}" virtual-host="${rabbitmq.virtual-host}"/> <!--定义管理交换机、队列--> <rabbit:admin connection-factory="connectionFactory"/> <!--定义持久化队列，不存在则自动创建；不绑定到交换机则绑定到默认交换机 默认交换机类型为direct，名字为：""，路由键为队列的名称 --> <rabbit:queue id="spring_queue" name="spring_queue" auto-declare="true"/> <!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~广播；所有队列都能收到消息~~~~~~~~~~~~~~~~~~~~~~~~~~~~ --> <!--定义广播交换机中的持久化队列，不存在则自动创建--> <rabbit:queue id="spring_fanout_queue_1" name="spring_fanout_queue_1" auto-declare="true"/> <!--定义广播交换机中的持久化队列，不存在则自动创建--> <rabbit:queue id="spring_fanout_queue_2" name="spring_fanout_queue_2" auto-declare="true"/> <!--定义广播类型交换机；并绑定上述两个队列--> <rabbit:fanout-exchange id="spring_fanout_exchange" name="spring_fanout_exchange" auto-declare="true"> <rabbit:bindings> <rabbit:binding queue="spring_fanout_queue_1"/> <rabbit:binding queue="spring_fanout_queue_2"/> </rabbit:bindings> </rabbit:fanout-exchange> <!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~通配符；*匹配一个单词，#匹配多个单词 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~ --> <!--定义广播交换机中的持久化队列，不存在则自动创建--> <rabbit:queue id="spring_topic_queue_star" name="spring_topic_queue_star" auto-declare="true"/> <!--定义广播交换机中的持久化队列，不存在则自动创建--> <rabbit:queue id="spring_topic_queue_well" name="spring_topic_queue_well" auto-declare="true"/> <!--定义广播交换机中的持久化队列，不存在则自动创建--> <rabbit:queue id="spring_topic_queue_well2" name="spring_topic_queue_well2" auto-declare="true"/> <rabbit:topic-exchange id="spring_topic_exchange" name="spring_topic_exchange" auto-declare="true"> <rabbit:bindings> <rabbit:binding pattern="cf.*" queue="spring_topic_queue_star"/> <rabbit:binding pattern="title.#" queue="spring_topic_queue_well"/> <rabbit:binding pattern="item.#" queue="spring_topic_queue_well2"/> </rabbit:bindings> </rabbit:topic-exchange> <!--定义rabbitTemplate对象操作可以在代码中方便发送消息--> <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/> </beans>3 发送消息
@RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration(locations = "classpath:spring-rabbitmq.xml") public class ProducerTest { @Autowired private RabbitTemplate rabbitTemplate; /** * 只发队列消息 * 默认交换机类型为 direct * 交换机的名称为空，路由键为队列的名称 */ @Test public void queueTest(){ //路由键与队列同名 rabbitTemplate.convertAndSend("spring_queue", "只发队列spring_queue的消息。"); } /** * 发送广播 * 交换机类型为 fanout * 绑定到该交换机的所有队列都能够收到消息 */ @Test public void fanoutTest(){ /** * 参数1：交换机名称 * 参数2：路由键名（广播设置为空） * 参数3：发送的消息内容 */ rabbitTemplate.convertAndSend("spring_fanout_exchange", "", "发送到spring_fanout_exchange交换机的广播消息"); } /** * 通配符 * 交换机类型为 topic * 匹配路由键的通配符，*表示一个单词，#表示多个单词 * 绑定到该交换机的匹配队列能够收到对应消息 */ @Test public void topicTest(){ /** * 参数1：交换机名称 * 参数2：路由键名 * 参数3：发送的消息内容 */ rabbitTemplate.convertAndSend("spring_topic_exchange", "cf.bj", "发送到spring_topic_exchange交换机heima.bj的消息"); rabbitTemplate.convertAndSend("spring_topic_exchange", "title.bj.1", "发送到spring_topic_exchange交换机heima.bj.1的消息"); rabbitTemplate.convertAndSend("spring_topic_exchange", "title.bj.2", "发送到spring_topic_exchange交换机heima.bj.2的消息"); rabbitTemplate.convertAndSend("spring_topic_exchange", "item.cn", "发送到spring_topic_exchange交换机itcast.cn的消息"); } }1 在pom.xml配置文件中添加依赖
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.cf</groupId> <artifactId>spring-rabbitmq-consumer</artifactId> <version>1.0-SNAPSHOT</version> <dependencies> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-context</artifactId> <version>5.1.7.RELEASE</version> </dependency> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit</artifactId> <version>2.1.8.RELEASE</version> </dependency> </dependencies> </project>2.1 创建rabbitmq.properties配置文件
rabbitmq.host=192.168.174.90 rabbitmq.port=5672 rabbitmq.username=cf rabbitmq.password=12345678 rabbitmq.virtual-host=/cf2.2 创建spring-rabbitmq.xml整合配置文件
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xmlns:rabbit="http://www.springframework.org/schema/rabbit" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context https://www.springframework.org/schema/context/spring-context.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd"> <!--加载配置文件--> <context:property-placeholder location="classpath:rabbitmq.properties"/> <!-- 定义rabbitmq connectionFactory --> <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}" port="${rabbitmq.port}" username="${rabbitmq.username}" password="${rabbitmq.password}" virtual-host="${rabbitmq.virtual-host}"/> <bean id="springQueueListener" class="com.cf.rabbitmq.listener.SpringQueueListener"/> <bean id="fanoutListener1" class="com.cf.rabbitmq.listener.FanoutListener1"/> <bean id="fanoutListener2" class="com.cf.rabbitmq.listener.FanoutListener2"/> <bean id="topicListenerStar" class="com.cf.rabbitmq.listener.TopicListenerStar"/> <bean id="topicListenerWell" class="com.cf.rabbitmq.listener.TopicListenerWell"/> <bean id="topicListenerWell2" class="com.cf.rabbitmq.listener.TopicListenerWell2"/> <rabbit:listener-container connection-factory="connectionFactory" auto-declare="true"> <rabbit:listener ref="springQueueListener" queue-names="spring_queue"/> <rabbit:listener ref="fanoutListener1" queue-names="spring_fanout_queue_1"/> <rabbit:listener ref="fanoutListener2" queue-names="spring_fanout_queue_2"/> <rabbit:listener ref="topicListenerStar" queue-names="spring_topic_queue_star"/> <rabbit:listener ref="topicListenerWell" queue-names="spring_topic_queue_well"/> <rabbit:listener ref="topicListenerWell2" queue-names="spring_topic_queue_well2"/> </rabbit:listener-container> </beans>3 创建消息监听器 1 队列监听器
public class SpringQueueListener implements MessageListener { public void onMessage(Message message) { try { String msg = new String(message.getBody(), "utf-8"); System.out.printf("接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n", message.getMessageProperties().getReceivedExchange(), message.getMessageProperties().getReceivedRoutingKey(), message.getMessageProperties().getConsumerQueue(), msg); } catch (Exception e) { e.printStackTrace(); } } }2 广播监听器1
public class FanoutListener1 implements MessageListener { public void onMessage(Message message) { try { String msg = new String(message.getBody(), "utf-8"); System.out.printf("广播监听器1：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n", message.getMessageProperties().getReceivedExchange(), message.getMessageProperties().getReceivedRoutingKey(), message.getMessageProperties().getConsumerQueue(), msg); } catch (Exception e) { e.printStackTrace(); } } }3 广播监听器2
public class FanoutListener2 implements MessageListener { public void onMessage(Message message) { try { String msg = new String(message.getBody(), "utf-8"); System.out.printf("广播监听器2：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n", message.getMessageProperties().getReceivedExchange(), message.getMessageProperties().getReceivedRoutingKey(), message.getMessageProperties().getConsumerQueue(), msg); } catch (Exception e) { e.printStackTrace(); } } }4 星号通配符监听器
public class TopicListenerStar implements MessageListener { public void onMessage(Message message) { try { String msg = new String(message.getBody(), "utf-8"); System.out.printf("通配符*监听器：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n", message.getMessageProperties().getReceivedExchange(), message.getMessageProperties().getReceivedRoutingKey(), message.getMessageProperties().getConsumerQueue(), msg); } catch (Exception e) { e.printStackTrace(); } } }5 井号通配符监听器
public class TopicListenerWell implements MessageListener { public void onMessage(Message message) { try { String msg = new String(message.getBody(), "utf-8"); System.out.printf("通配符#监听器：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n", message.getMessageProperties().getReceivedExchange(), message.getMessageProperties().getReceivedRoutingKey(), message.getMessageProperties().getConsumerQueue(), msg); } catch (Exception e) { e.printStackTrace(); } } }6 井号通配符监听器2
public class TopicListenerWell2 implements MessageListener { public void onMessage(Message message) { try { String msg = new String(message.getBody(), "utf-8"); System.out.printf("通配符#监听器2：接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n", message.getMessageProperties().getReceivedExchange(), message.getMessageProperties().getReceivedRoutingKey(), message.getMessageProperties().getConsumerQueue(), msg); } catch (Exception e) { e.printStackTrace(); } } }步骤分析: 生产者工程：
application.yml文件配置RabbitMQ相关信息在生产者工程中编写配置类，用于创建交换机和队列，并进行绑定注入RabbitTemplate对象，通过RabbitTemplate对象发送消息到交换机消费者工程：
application.yml文件配置RabbitMQ相关信息创建消息处理类，用于接收队列中的消息并进行处理创建生产者工程springboot-rabbitmq-producer: 1 pom.xml添加依赖
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.1.4.RELEASE</version> </parent> <groupId>com.cf</groupId> <artifactId>springboot-rabbitmq-producer</artifactId> <version>1.0-SNAPSHOT</version> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> </dependency> </dependencies> </project>2 启动类
package com.cf.rabbitmq; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class ProducerApplication { public static void main(String[] args) { SpringApplication.run(ProducerApplication.class); } }3 创建application.yml
spring: rabbitmq: host: localhost port: 5672 virtual-host: /cf username: cf password: 123456784 创建RabbitMQ队列与交换机绑定的配置类
package com.cf.rabbitmq.config; import org.springframework.amqp.core.*; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitMQConfig { //交换机名称 public static final String ITEM_TOPIC_EXCHANGE = "item_topic_exchange"; //队列名称 public static final String ITEM_QUEUE = "item_queue"; //声明交换机 @Bean("itemTopicExchange") public Exchange topicExchange(){ return ExchangeBuilder.topicExchange(ITEM_TOPIC_EXCHANGE).durable(true).build(); } //声明队列 @Bean("itemQueue") public Queue itemQueue(){ return QueueBuilder.durable(ITEM_QUEUE).build(); } //绑定队列和交换机 @Bean public Binding itemQueueExchange(@Qualifier("itemQueue") Queue queue, @Qualifier("itemTopicExchange") Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with("item.#").noargs(); } }创建消费者工程springboot-rabbitmq-consumer: 1 pom.xml添加依赖
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.1.4.RELEASE</version> </parent> <groupId>com.cf</groupId> <artifactId>springboot-rabbitmq-consumer</artifactId> <version>1.0-SNAPSHOT</version> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> </dependencies> </project>2 启动类
package com.cf.rabbitmq; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class ConsumerApplication { public static void main(String[] args) { SpringApplication.run(ConsumerApplication.class); } }3 创建application.yml
spring: rabbitmq: host: localhost port: 5672 virtual-host: /cf username: cf password: 123456784 消息监听处理类
package com.cf.rabbitmq.listener; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component public class MyListener { /** * 监听某个队列的消息 * @param message 接收到的消息 */ @RabbitListener(queues = "item_queue") public void myListener1(String message){ System.out.println("消费者接收到的消息为：" + message); } }5 测试 在生产者工程springboot-rabbitmq-producer中创建测试类，发送消息:
package com.cf.rabbitmq; import com.cf.rabbitmq.config.RabbitMQConfig; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; @RunWith(SpringRunner.class) @SpringBootTest public class RabbitMQTest { @Autowired private RabbitTemplate rabbitTemplate; @Test public void test(){ rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE, "item.insert", "商品新增，routing key 为item.insert"); rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE, "item.update", "商品修改，routing key 为item.update"); rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE, "item.delete", "商品删除，routing key 为item.delete"); } }注意:先运行上述测试程序（交换机和队列才能先被声明和绑定），然后启动消费者；在消费者工程springboot-rabbitmq-consumer中控制台查看是否接收到对应消息.
