- 个人主页:陶然同学
- 版权:本文由【陶然同学】原创、在CSDN首发、需要转载请联系博主
- 如果文章对你有帮助、欢迎关注、点赞、收藏(一键三连)和订阅专栏哦
- 想寻找共同成长的小伙伴，请点击【Java全栈开发社区】
目录
1.MQ消息队列
1.1什么是MQ
1.2实现方式：AMQP、JMS
1.3常见MQ产品
2.RabbitMQ
2.1RabbitMQ概述
2.1.1什么是Rabbit
2.1.2RabbitMQ的工作原理
2.1.3消息流程
2.2下载和安装
2.2.1下载
2.2.2安装
2.2.3运行
2.2.4访问
2.3入门案例
2.3.1环境搭建
2.3.2生产者
2.3.3消费者
2.4工作模式
2.4.1Word queues
2.4.2Publish/Subscribe
2.4.3Routing
2.4.4Topic
2.4.5Header
2.4.6RPC
2.5SpringBoot整合RabbitMQ
2.5.1搭建环境
2.5.2配置类
2.5.3生产者
2.5.4消费者
3.写在最后
1.MQ消息队列
1.1什么是MQ
消息队列，即MQ，Message Queue。
- 消息队列是典型的：生产者、消费者模型。
- 生产者不断向消息队列中生产消息，
- 消费者不断的从队列中获取消息。
- 因为消息的生产和消费都是异步的
而且只关心消息的发送和接收，没有业务逻辑的侵入，这样就实现了生产者和消费者的解耦。
1.2实现方式：AMQP、JMS
MQ是消息通信的模型，并不是具体实现。现在实现MQ的有两种主流方式：AMQP、JMS。
- 两者间的区别和联系：
- JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式
- JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
- JMS规定了两种消息模型；而AMQP的消息模型更加丰富
1.3常见MQ产品
- ActiveMQ：基于JMS
- RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好
- RocketMQ：基于JMS，阿里巴巴产品，目前交由Apache基金会
- Kafka：分布式消息系统，高吞吐量
2.RabbitMQ
2.1RabbitMQ概述
2.1.1什么是Rabbit
MQ全称为Message Queue，即消息队列， RabbitMQ是由erlang语言开发，基于AMQP（Advanced Message Queue 高级消息队列协议）协议实现的消息队列，它是一种应用程序之间的通信方法，消息队列在分布式系统开发中应用非常广泛。
RabbitMQ官方地址：Messaging that just works — RabbitMQ
快速入门：RabbitMQ Tutorials — RabbitMQ
2.1.2RabbitMQ的工作原理
组成部分说明如下
- Broker：消息队列服务进程，此进程包括两个部分：Exchange和Queue。
- Exchange：消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过虑。
- Queue：消息队列，存储消息的队列，消息到达队列并转发给指定的消费方。
- Producer：消息生产者，即生产方客户端，生产方客户端将消费发送到MQ。
- Consumer：消息消费者，即消费方客户端，接收MQ转发的消息。
2.1.3消息流程
- 发送消息流程
1、生产者和Broker建立TCP连接。
2、生产者和Broker建立通道。
3、生产者通过通道消息发送Broker，由Exchange将消息进行转发。
4、Exchange将消息转发到指定的Queue（队列）
- 接收消息流程
1、消费者和Broker建立TCP连接
2、消费者和Broker建立通道
3、消费者监听指定的Queue（队列）
4、当有消息到达Queue时Broker默认将消息推送给消费者。
5、消费者接收到消息。
2.2下载和安装
2.2.1下载
- 官网下载地址：Downloading and Installing RabbitMQ — RabbitMQ
- https://www.rabbitmq.com/install-windows.html#installer
- 目前最新版本是：3.8.3
2.2.2安装
- 安装erlang
- 安装RabbitMQ
2.2.3运行
- 步骤一：打开控制台
- 步骤二：激活插件
rabbitmq-plugins.bat enable rabbitmq_management
- 步骤三：重启
net stop rabbitmq && net start rabbitmq
2.2.4访问
- 步骤一：输入 http://127.0.0.1:15672/
- 步骤二：默认账号登录：guest/ guest
2.3入门案例
2.3.1环境搭建
- 构建环境：test_rabbitmq
org.springframework.boot spring-boot-starter-amqp 2.3.2生产者
ackage com.czxy.xuecheng.rabbitmq;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;import java.util.concurrent.TimeoutException;/** * Created by liangtong. */public class Producer01 { //队列名称 private static final String QUEUE = "helloworld"; public static void main(String[] args) throws IOException, TimeoutException { Connection connection = null; Channel channel = null; try { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); factory.setUsername("guest"); factory.setPassword("guest"); factory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器 // 创建与RabbitMQ服务的TCP连接 connection = factory.newConnection(); //创建与Exchange的通道，每个连接可以创建多个通道，每个通道代表一个会话任务 channel = connection.createChannel(); /** * 声明队列，如果Rabbit中没有此队列将自动创建 * param1:队列名称 * param2:是否持久化 * param3:队列是否独占此连接 * param4:队列不再使用时是否自动删除此队列 * param5:队列参数 */ channel.queueDeclare(QUEUE, true, false, false, null); String message = "helloworld小明"+System.currentTimeMillis(); /** * 消息发布方法 * param1：Exchange的名称，如果没有指定，则使用Default Exchange * param2:routingKey,消息的路由Key，是用于Exchange（交换机）将消息转发到指定的消息队列 * param3:消息包含的属性 * param4：消息体 */ /** * 这里没有指定交换机，消息将发送给默认交换机，每个队列也会绑定那个默认的交换机，但是不能显 示绑定或解除绑定 * 默认的交换机，routingKey等于队列名称 */ channel.basicPublish("", QUEUE, null, message.getBytes()); System.out.println("Send Message is:'" + message + "'"); } catch(Exception ex) { ex.printStackTrace(); } finally { if(channel != null) { channel.close(); } if(connection != null) { connection.close(); } } }}2.3.3消费者
package com.czxy.xuecheng.rabbitmq;import com.rabbitmq.client.*;import java.io.IOException;import java.util.concurrent.TimeoutException;/** * Created by liangtong. */public class Consumer01 { private static final String QUEUE = "helloworld"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); //设置MabbitMQ所在服务器的ip和端口 factory.setHost("127.0.0.1"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); //声明队列 channel.queueDeclare(QUEUE, true, false, false, null); //定义消费方法 DefaultConsumer consumer = new DefaultConsumer(channel) { /** * 消费者接收消息调用此方法 * @param consumerTag 消费者的标签，在channel.basicConsume()去指定 * @param envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送) * @param properties * @param body * @throws IOException */ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { //交换机 String exchange = envelope.getExchange(); //路由key String routingKey = envelope.getRoutingKey(); //消息id long deliveryTag = envelope.getDeliveryTag(); //消息内容 String msg = new String(body,"utf-8"); System.out.println("receive message.." + msg); } }; /** * 监听队列String queue, boolean autoAck,Consumer callback * 参数明细 * 1、队列名称 * 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动回复 * 3、消费消息的方法，消费者接收到消息后调用此方法 */ channel.basicConsume(QUEUE, true, consumer); }}2.4工作模式
RabbitMQ有以下几种工作模式 ：
1、Work queues
2、Publish/Subscribe：发布订阅模式
3、Routing：路由模式
4、Topics
5、Header
6、RPC
2.4.1Word queues
work queues与入门程序相比，多了一个消费端，两个消费端共同消费同一个队列中的消息。
- *测试：
1、使用入门程序，启动多个消费者。
2、生产者发送多个消息。
- 结果：
1、一条消息只会被一个消费者接收；
2、rabbit采用轮询的方式将消息是平均发送给消费者的；
3、消费者在处理完某条消息后，才会收到下一条消息。
2.4.2Publish/Subscribe
发布订阅模式：
1、每个消费者监听自己的队列。
2、生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息
- 需求：用户通知，当用户充值成功或转账完成系统通知用户，通知方式有短信、邮件多种方法 。
- 1）生产者
声明inform_Exchange_fanout交换机。
声明两个队列并且绑定到此交换机，绑定时不需要指定routingkey
发送消息时不需要指定routingkey
package com.czxy.xuecheng.rabbitmq.demo02;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;/** * Created by liangtong. */public class Producer02_publish { //队列名称 private static final String QUEUE_INFORM_EMAIL = "queue_inform_email"; private static final String QUEUE_INFORM_SMS = "queue_inform_sms"; private static final String EXCHANGE_FANOUT_INFORM = "inform_exchange_fanout"; public static void main(String[] args) { Connection connection = null; Channel channel = null; try { //创建一个与MQ的连接 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); factory.setPort(5672); factory.setUsername("guest"); factory.setPassword("guest"); factory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器 //创建一个连接 connection = factory.newConnection(); //创建与交换机的通道，每个通道代表一个会话 channel = connection.createChannel(); //声明交换机 String exchange, BuiltinExchangeType type /** * 参数明细 * 1、交换机名称 * 2、交换机类型，fanout、topic、direct、headers * fanout：广播模式需要生产者消费者绑定相同的Exchange * topic：支持模糊匹配的广播模式以点分隔，*表示一个单词，#表示任意数量（零个或多个）单词 * direct ：需要生产者和消费者绑定相同的Exchange和routing key * headers：根据生产者和消费者的header中信息进行匹配性能较差 ，x-match [all 匹配所有/any 任意一个]。 */ channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT); //声明队列 // (String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments) /** * 参数明细： * 1、队列名称 * 2、是否持久化 * 3、是否独占此队列 * 4、队列不用是否自动删除 * 5、参数 */ channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null); channel.queueDeclare(QUEUE_INFORM_SMS, true, false, false, null); //交换机和队列绑定String queue, String exchange, String routingKey /** * 参数明细 * 1、队列名称 * 2、交换机名称 * 3、路由key */ channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_FANOUT_INFORM, ""); channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_FANOUT_INFORM, ""); //发送消息 for (int i = 0; i < 10; i++) { String message = "inform to user" + i; //向交换机发送消息 String exchange, String routingKey, BasicProperties props, byte[] body /** * 参数明细 * 1、交换机名称，不指令使用默认交换机名称 Default Exchange * 2、routingKey（路由key），根据key名称将消息转发到具体的队列，这里填写队列名称表示消息将发到此队列 * 3、消息属性 * 4、消息内容 */ channel.basicPublish(EXCHANGE_FANOUT_INFORM, "", null, message.getBytes()); System.out.println("Send Message is:'" + message + "'"); } } catch (Exception e) { e.printStackTrace(); } finally { if (channel != null) { try { channel.close(); } catch (Exception e) { e.printStackTrace(); } } if (connection != null) { try { connection.close(); } catch (IOException e) { e.printStackTrace(); } } } }}- 2）邮件发送消费者
package com.czxy.xuecheng.rabbitmq.demo02;import com.rabbitmq.client.*;import java.io.IOException;/** * Created by liangtong. */public class Consumer02_subscribe_email { //队列名称 private static final String QUEUE_INFORM_EMAIL = "queue_inform_email"; private static final String EXCHANGE_FANOUT_INFORM = "inform_exchange_fanout"; public static void main(String[] args) throws Exception { //创建一个与MQ的连接 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); factory.setPort(5672); factory.setUsername("guest"); factory.setPassword("guest"); factory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器 //创建一个连接 Connection connection = factory.newConnection(); //创建与交换机的通道，每个通道代表一个会话 Channel channel = connection.createChannel(); //声明交换机 String exchange, BuiltinExchangeType type /** * 参数明细 * 1、交换机名称 * 2、交换机类型，fanout、topic、direct、headers */ channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT); //声明队列 //channel.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments) /** * 参数明细： * 1、队列名称 * 2、是否持久化 * 3、是否独占此队列 * 4、队列不用是否自动删除 * 5、参数 */ channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null); //交换机和队列绑定String queue, String exchange, String routingKey /** * 参数明细 * 1、队列名称 * 2、交换机名称 * 3、路由key */ channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_FANOUT_INFORM, ""); //定义消费方法 DefaultConsumer defaultConsumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { long deliveryTag = envelope.getDeliveryTag(); String exchange = envelope.getExchange(); //消息内容 String message = new String(body, "UTF-8"); System.out.println(message); } }; /** * 监听队列String queue, boolean autoAck,Consumer callback * 参数明细 * 1、队列名称 * 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置 为false则需要手动回复 * 3、消费消息的方法，消费者接收到消息后调用此方法 */ channel.basicConsume(QUEUE_INFORM_EMAIL, true, defaultConsumer); }}- 3）短信发送消费者
package com.czxy.xuecheng.rabbitmq.demo02;import com.rabbitmq.client.*;import java.io.IOException;/** * Created by liangtong. */public class Consumer02_subscribe_sms { //队列名称 private static final String QUEUE_INFORM_SMS = "queue_inform_sms"; private static final String EXCHANGE_FANOUT_INFORM = "inform_exchange_fanout"; public static void main(String[] args) throws Exception { //创建一个与MQ的连接 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); factory.setPort(5672); factory.setUsername("guest"); factory.setPassword("guest"); //rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器 factory.setVirtualHost("/"); //创建一个连接 Connection connection = factory.newConnection(); //创建与交换机的通道，每个通道代表一个会话 Channel channel = connection.createChannel(); //声明交换机 String exchange, BuiltinExchangeType type /** * 参数明细 * 1、交换机名称 * 2、交换机类型，fanout、topic、direct、headers */ channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT); //声明队列 //channel.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments) /** * 参数明细： * 1、队列名称 * 2、是否持久化 * 3、是否独占此队列 * 4、队列不用是否自动删除 * 5、参数 */ channel.queueDeclare(QUEUE_INFORM_SMS, true, false, false, null); //交换机和队列绑定String queue, String exchange, String routingKey /** * 参数明细 * 1、队列名称 * 2、交换机名称 * 3、路由key */ channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_FANOUT_INFORM, ""); //定义消费方法 DefaultConsumer defaultConsumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { long deliveryTag = envelope.getDeliveryTag(); String exchange = envelope.getExchange(); //消息内容 String message = new String(body, "UTF-8"); System.out.println(message); } }; /** * 监听队列String queue, boolean autoAck,Consumer callback * 参数明细 * 1、队列名称 * 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置 为false则需要手动回复 * 3、消费消息的方法，消费者接收到消息后调用此方法 */ channel.basicConsume(QUEUE_INFORM_SMS, true, defaultConsumer); }}2.4.3Routing
路由模式：
1、每个消费者监听自己的队列，并且设置routingkey。
2、生产者将消息发给交换机，由交换机根据routingkey来转发消息到指定的队列。
- 1）生产者
声明inform_exchange_routing交换机。
声明两个队列并且绑定到此交换机，绑定时需要指定routingkey
发送消息时需要指定routingkey
package com.czxy.xuecheng.rabbitmq.demo03;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;/** * Created by liangtong. */public class Producer03_routing { //队列名称 private static final String QUEUE_INFORM_EMAIL = "queue_inform_email"; private static final String QUEUE_INFORM_SMS = "queue_inform_sms"; private static final String EXCHANGE_ROUTING_INFORM = "inform_exchange_routing"; public static void main(String[] args) { Connection connection = null; Channel channel = null; try { //创建一个与MQ的连接 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); factory.setPort(5672); factory.setUsername("guest"); factory.setPassword("guest"); //rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器 factory.setVirtualHost("/"); //创建一个连接 connection = factory.newConnection(); //创建与交换机的通道，每个通道代表一个会话 channel = connection.createChannel(); //声明交换机 String exchange, BuiltinExchangeType type /** * 参数明细 * 1、交换机名称 * 2、交换机类型，fanout、topic、direct、headers */ channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT); //声明队列 //channel.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments) /** * 参数明细： * 1、队列名称 * 2、是否持久化 * 3、是否独占此队列 * 4、队列不用是否自动删除 * 5、参数 */ channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null); channel.queueDeclare(QUEUE_INFORM_SMS, true, false, false, null); //交换机和队列绑定String queue, String exchange, String routingKey /** * 参数明细 * 1、队列名称 * 2、交换机名称 * 3、路由key */ channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_ROUTING_INFORM, QUEUE_INFORM_EMAIL); channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_ROUTING_INFORM, QUEUE_INFORM_SMS); //发送邮件消息 for (int i = 0; i < 10; i++) { String message = "email inform to user" + i; //向交换机发送消息 String exchange, String routingKey, BasicProperties props,byte[] body /** * 参数明细 * 1、交换机名称，不指令使用默认交换机名称 Default Exchange * 2、routingKey（路由key），根据key名称将消息转发到具体的队列，这里填写队列名称表示消 息将发到此队列 * 3、消息属性 * 4、消息内容 */ channel.basicPublish(EXCHANGE_ROUTING_INFORM, QUEUE_INFORM_EMAIL, null, message.getBytes()); System.out.println("Send Message is:'" + message + "'"); } //发送短信消息 for (int i = 0; i < 10; i++) { String message = "sms inform to user" + i; //向交换机发送消息 String exchange, String routingKey, BasicProperties props,byte[] body channel.basicPublish(EXCHANGE_ROUTING_INFORM, QUEUE_INFORM_SMS, null, message.getBytes()); System.out.println("Send Message is:'" + message + "'"); } } catch (Exception e) { e.printStackTrace(); } finally { if (channel != null) { try { channel.close(); } catch (Exception e) { e.printStackTrace(); } } if (connection != null) { try { connection.close(); } catch (IOException e) { e.printStackTrace(); } } } }}- 2）邮件发送消费者
package com.czxy.xuecheng.rabbitmq.demo03;import com.rabbitmq.client.*;import java.io.IOException;/** * Created by liangtong. */public class Consumer03_routing_email { //队列名称 private static final String QUEUE_INFORM_EMAIL = "queue_inform_email"; private static final String EXCHANGE_ROUTING_INFORM = "inform_exchange_routing"; public static void main(String[] args) throws Exception { //创建一个与MQ的连接 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); factory.setPort(5672); factory.setUsername("guest"); factory.setPassword("guest"); //rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器 factory.setVirtualHost("/"); //创建一个连接 Connection connection = factory.newConnection(); //创建与交换机的通道，每个通道代表一个会话 Channel channel = connection.createChannel(); //声明交换机 String exchange, BuiltinExchangeType type /** * 参数明细 * 1、交换机名称 * 2、交换机类型，fanout、topic、direct、headers */ channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT); //声明队列 //channel.queueDeclare(String queue, boolean durable, boolean exclusive, booleanautoDelete, Map arguments) /** * 参数明细： * 1、队列名称 * 2、是否持久化 * 3、是否独占此队列 * 4、队列不用是否自动删除 * 5、参数 */ channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null); //交换机和队列绑定String queue, String exchange, String routingKey /** * 参数明细 * 1、队列名称 * 2、交换机名称 * 3、路由key */ channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_ROUTING_INFORM, QUEUE_INFORM_EMAIL); //定义消费方法 DefaultConsumer defaultConsumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { long deliveryTag = envelope.getDeliveryTag(); String exchange = envelope.getExchange(); //消息内容 String message = new String(body, "UTF-8"); System.out.println(message); } }; /** * 监听队列String queue, boolean autoAck,Consumer callback * 参数明细 * 1、队列名称 * 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置 为false则需要手动回复 * 3、消费消息的方法，消费者接收到消息后调用此方法 */ channel.basicConsume(QUEUE_INFORM_EMAIL, true, defaultConsumer); }}- 3）短信发送消费者
参考邮件发送消费者的代码流程，编写短信通知的代码。
package com.czxy.xuecheng.rabbitmq.demo03;import com.rabbitmq.client.*;import java.io.IOException;/** * Created by liangtong. */public class Consumer03_routing_sms { //队列名称 private static final String QUEUE_INFORM_SMS = "queue_inform_sms"; private static final String EXCHANGE_ROUTING_INFORM = "inform_exchange_routing"; public static void main(String[] args) throws Exception { //创建一个与MQ的连接 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); factory.setPort(5672); factory.setUsername("guest"); factory.setPassword("guest"); //rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器 factory.setVirtualHost("/"); //创建一个连接 Connection connection = factory.newConnection(); //创建与交换机的通道，每个通道代表一个会话 Channel channel = connection.createChannel(); //声明交换机 String exchange, BuiltinExchangeType type /** * 参数明细 * 1、交换机名称 * 2、交换机类型，fanout、topic、direct、headers */ channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT); //声明队列 //channel.queueDeclare(String queue, boolean durable, boolean exclusive, booleanautoDelete, Map arguments) /** * 参数明细： * 1、队列名称 * 2、是否持久化 * 3、是否独占此队列 * 4、队列不用是否自动删除 * 5、参数 */ channel.queueDeclare(QUEUE_INFORM_SMS, true, false, false, null); //交换机和队列绑定String queue, String exchange, String routingKey /** * 参数明细 * 1、队列名称 * 2、交换机名称 * 3、路由key */ channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_ROUTING_INFORM, QUEUE_INFORM_SMS); //定义消费方法 DefaultConsumer defaultConsumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { long deliveryTag = envelope.getDeliveryTag(); String exchange = envelope.getExchange(); //消息内容 String message = new String(body, "UTF-8"); System.out.println(message); } }; /** * 监听队列String queue, boolean autoAck,Consumer callback * 参数明细 * 1、队列名称 * 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置 为false则需要手动回复 * 3、消费消息的方法，消费者接收到消息后调用此方法 */ channel.basicConsume(QUEUE_INFORM_SMS, true, defaultConsumer); }}2.4.4Topic
- 主题模式
1、每个消费者监听自己的队列，并且设置带统配符的routingkey。
2、生产者将消息发给broker，由交换机根据routingkey来转发消息到指定的队列。
- 需求：根据用户的通知设置去通知用户，设置接收Email的用户只接收Email，设置接收sms的用户只接收sms，设置两种通知类型都接收的则两种通知都有效。
- 1）生产者
队列绑定交换机指定通配符：
统配符规则：
中间以“.”分隔。
符号#可以匹配多个词，符号*可以匹配一个词语。
package com.czxy.xuecheng.rabbitmq.demo04;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;/** * Created by liangtong. */public class Producer04_topic { //队列名称 private static final String QUEUE_INFORM_EMAIL = "queue_inform_email"; private static final String QUEUE_INFORM_SMS = "queue_inform_sms"; private static final String EXCHANGE_TOPIC_INFORM = "inform_exchange_topic"; public static void main(String[] args) { Connection connection = null; Channel channel = null; try { //创建一个与MQ的连接 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); factory.setPort(5672); factory.setUsername("guest"); factory.setPassword("guest"); //rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器 factory.setVirtualHost("/"); //创建一个连接 connection = factory.newConnection(); //创建与交换机的通道，每个通道代表一个会话 channel = connection.createChannel(); //声明交换机 String exchange, BuiltinExchangeType type /** * 参数明细 * 1、交换机名称 * 2、交换机类型，fanout、topic、direct、headers */ channel.exchangeDeclare(EXCHANGE_TOPIC_INFORM, BuiltinExchangeType.TOPIC); //声明队列 //channel.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map arguments) /** * 参数明细： * 1、队列名称 * 2、是否持久化 * 3、是否独占此队列 * 4、队列不用是否自动删除 * 5、参数 */ channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null); channel.queueDeclare(QUEUE_INFORM_SMS, true, false, false, null); //交换机和队列绑定String queue, String exchange, String routingKey /** * 参数明细 * 1、队列名称 * 2、交换机名称 * 3、路由key */ channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_TOPIC_INFORM, "inform.#.email.#"); channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_TOPIC_INFORM, "inform.#.sms.#"); //发送邮件消息 for (int i = 0; i < 10; i++) { String message = "email inform to user" + i; //向交换机发送消息 String exchange, String routingKey, BasicProperties props,byte[] body /** * 参数明细 * 1、交换机名称，不指令使用默认交换机名称 Default Exchange * 2、routingKey（路由key），根据key名称将消息转发到具体的队列，这里填写队列名称表示消 息将发到此队列 * 3、消息属性 * 4、消息内容 */ channel.basicPublish(EXCHANGE_TOPIC_INFORM, "inform.email", null, message.getBytes()); System.out.println("Send Message is:'" + message + "'"); } //发送短信消息 for (int i = 0; i < 10; i++) { String message = "sms inform to user" + i; //向交换机发送消息 String exchange, String routingKey, BasicProperties props,byte[] body channel.basicPublish(EXCHANGE_TOPIC_INFORM, "inform.sms", null, message.getBytes()); System.out.println("Send Message is:'" + message + "'"); } //发送邮件和短信消息 for (int i = 0; i < 10; i++) { String message = "sms and email inform to user" + i; //向交换机发送消息 String exchange, String routingKey, BasicProperties props,byte[] body channel.basicPublish(EXCHANGE_TOPIC_INFORM, "inform.sms.email", null, message.getBytes()); System.out.println("Send Message is:'" + message + "'"); } } catch (Exception e) { e.printStackTrace(); } finally { if (channel != null) { try { channel.close(); } catch (Exception e) { e.printStackTrace(); } } if (connection != null) { try { connection.close(); } catch (IOException e) { e.printStackTrace(); } } } }}- 2）消费者
package com.czxy.xuecheng.rabbitmq.demo04;import com.rabbitmq.client.*;import java.io.IOException;/** * Created by liangtong. */public class Consumer04_topic_email { //队列名称 private static final String QUEUE_INFORM_EMAIL = "queue_inform_email"; private static final String EXCHANGE_TOPIC_INFORM = "inform_exchange_topic"; public static void main(String[] args) throws Exception { //创建一个与MQ的连接 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); factory.setPort(5672); factory.setUsername("guest"); factory.setPassword("guest"); //rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务器 factory.setVirtualHost("/"); //创建一个连接 Connection connection = factory.newConnection(); //创建与交换机的通道，每个通道代表一个会话 Channel channel = connection.createChannel(); //声明交换机 String exchange, BuiltinExchangeType type /** * 参数明细 * 1、交换机名称 * 2、交换机类型，fanout、topic、direct、headers */ channel.exchangeDeclare(EXCHANGE_TOPIC_INFORM, BuiltinExchangeType.TOPIC); //声明队列 //channel.queueDeclare(String queue, boolean durable, boolean exclusive, booleanautoDelete, Map arguments) /** * 参数明细： * 1、队列名称 * 2、是否持久化 * 3、是否独占此队列 * 4、队列不用是否自动删除 * 5、参数 */ channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null); //交换机和队列绑定String queue, String exchange, String routingKey /** * 参数明细 * 1、队列名称 * 2、交换机名称 * 3、路由key */ channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_TOPIC_INFORM, QUEUE_INFORM_EMAIL); //定义消费方法 DefaultConsumer defaultConsumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { long deliveryTag = envelope.getDeliveryTag(); String exchange = envelope.getExchange(); //消息内容 String message = new String(body, "UTF-8"); System.out.println(message); } }; /** * 监听队列String queue, boolean autoAck,Consumer callback * 参数明细 * 1、队列名称 * 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置 为false则需要手动回复 * 3、消费消息的方法，消费者接收到消息后调用此方法 */ channel.basicConsume(QUEUE_INFORM_EMAIL, true, defaultConsumer); }}2.4.5Header
header模式与routing不同的地方在于，header模式取消routingkey，使用header中的 key/value（键值对）匹配队列。
- 1）生产者
Mapheaders_email=newHashtable();headers_email.put("inform_type","email");Mapheaders_sms=newHashtable();headers_sms.put("inform_type","sms");channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_HEADERS_INFORM,"",headers_email);channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_HEADERS_INFORM,"",headers_sms);- 2）发送邮件消费者
channel.exchangeDeclare(EXCHANGE_HEADERS_INFORM,BuiltinExchangeType.HEADERS);Mapheaders_email=newHashtable();headers_email.put("inform_email","email");//交换机和队列绑定channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_HEADERS_INFORM,"",headers_email);//指定消费队列channel.basicConsume(QUEUE_INFORM_EMAIL,true,consumer);2.4.6RPC
RPC即客户端远程调用服务端的方法 ，使用MQ可以实现RPC的异步调用，基于Direct交换机实现，流程如下：
1、客户端即是生产者就是消费者，向RPC请求队列发送RPC调用消息，同时监听RPC响应队列。
2、服务端监听RPC请求队列的消息，收到消息后执行服务端的方法，得到方法返回的结果
3、服务端将RPC方法 的结果发送到RPC响应队列
4、客户端（RPC调用方）监听RPC响应队列，接收到RPC调用结果。
2.5SpringBoot整合RabbitMQ
2.5.1搭建环境
- 创建测试项目：test_rabbitmq_boot
- 添加依赖
xc_test_parent com.czxy.xuecheng 1.0-SNAPSHOT 4.0.0 test_rabbitmq_boot org.springframework.boot spring-boot-starter-amqp org.springframework.boot spring-boot-starter-test - 添加yml文件
server: port: 8090spring: application: name: test_rabbitmq_producer rabbitmq: host: 127.0.0.1 port: 5672 username: guest passowrd: guest virtualHost: /- 创建启动类：TestRabbitMQBootApplication
package com.czxy.xuecheng;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;/** * Created by liangtong. */@SpringBootApplicationpublic class TestRabbitMQBootApplication { public static void main(String[] args) { SpringApplication.run(TestRabbitMQBootApplication.class, args); }}2.5.2配置类
package com.czxy.xuecheng.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;/** * Created by liangtong. */@Configurationpublic class RabbitConfig { // 交换机名称 public static final String EXCHANGE_TOPIC_INFORM = "inform_exchange_topic"; //队列名称 public static final String QUEUE_INFORM_EMAIL = "queue_inform_email"; public static final String QUEUE_INFORM_SMS = "queue_inform_sms"; /** 交换机配置 * ExchangeBuilder提供了fanout、direct、topic、header交换机类型的配置 * channel.exchangeDeclare(EXCHANGE_TOPIC_INFORM, BuiltinExchangeType.TOPIC); * @return */ @Bean(EXCHANGE_TOPIC_INFORM) public Exchange exchange_topic() { //durable(true)持久化，消息队列重启后交换机仍然存在 return ExchangeBuilder.topicExchange(EXCHANGE_TOPIC_INFORM).durable(true).build(); } /** * 声明队列 * channel.queueDeclare(QUEUE_INFORM_SMS, true, false, false, null); * @return */ @Bean(QUEUE_INFORM_SMS) public Queue queue_inform_sms(){ return new Queue(QUEUE_INFORM_SMS); } @Bean(QUEUE_INFORM_EMAIL) public Queue queue_inform_email(){ return new Queue(QUEUE_INFORM_EMAIL,true,false,false); } /** * 绑定队列到交换机 * channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_TOPIC_INFORM, "inform.#.email.#"); * @param queue * @param exchange * @return */ @Bean public Binding binding_queue_inform_sms(@Qualifier(QUEUE_INFORM_SMS) Queue queue, @Qualifier(EXCHANGE_TOPIC_INFORM) Exchange exchange) { return BindingBuilder.bind(queue).to(exchange).with("inform.#.sms.#").noargs(); } @Bean public Binding binding_queue_inform_email(@Qualifier(QUEUE_INFORM_EMAIL) Queue queue, @Qualifier(EXCHANGE_TOPIC_INFORM) Exchange exchange) { return BindingBuilder.bind(queue).to(exchange).with("inform.#.email.#").noargs(); }}2.5.3生产者
package com.czxy.xuecheng;import com.czxy.xuecheng.config.RabbitConfig;import org.junit.Test;import org.junit.runner.RunWith;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.boot.test.context.SpringBootTest;import org.springframework.test.context.junit4.SpringRunner;import javax.annotation.Resource;/** * Created by liangtong. */@RunWith(SpringRunner.class)@SpringBootTest(classes = TestRabbitMQBootApplication.class)public class Producer05Topic { @Resource private RabbitTemplate rabbitTemplate; @Test public void testSendEmail() { //channel.basicPublish(EXCHANGE_TOPIC_INFORM, "inform.email", null, message.getBytes()); for(int i = 0 ; i < 5 ; i ++) { String message = "email inform to user" + i; rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_TOPIC_INFORM,"inform.email",message); System.out.println("Send Message is:'" + message + "'"); } } @Test public void testSendSms() { //channel.basicPublish(EXCHANGE_TOPIC_INFORM, "inform.sms", null, message.getBytes()); for(int i = 0 ; i < 5 ; i ++) { String message = "sms inform to user" + i; rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_TOPIC_INFORM,"inform.sms",message); System.out.println("Send Message is:'" + message + "'"); } } @Test public void testSendSmsAndEmail() { //channel.basicPublish(EXCHANGE_TOPIC_INFORM, "inform.sms.email", null, message.getBytes()); for(int i = 0 ; i < 5 ; i ++) { String message = "sms and email inform to user" + i; rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_TOPIC_INFORM,"inform.sms.email",message); System.out.println("Send Message is:'" + message + "'"); } }}2.5.4消费者
package com.czxy.xuecheng.listener;import com.czxy.xuecheng.config.RabbitConfig;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;/** * Created by liangtong. */@Componentpublic class Consumer05Topic { @RabbitListener(queues = RabbitConfig.QUEUE_INFORM_EMAIL) public void receiveEmail(String msg , Message message){ System.out.println("receive message is:" + msg); }/* @RabbitListener(queues = RabbitConfig.QUEUE_INFORM_SMS) public void receiveSmS(String msg , Message message){ System.out.println("receive message is:" + msg); } */}