devbox
小小林熬夜学编程
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

RabbitMQ简介及其六种工作模式_mqtru

作者：小小林熬夜学编程 | 2024-05-19 21:24:18

赞

踩

mqtru

本文参考自：

https://blog.csdn.net/kavito/article/details/91403659

https://www.cnblogs.com/zpKang/p/13599813.html

什么是消息队列

MQ全称为Message Queue，即消息队列。“消息队列”是在消息的传输过程中保存消息的容器。它是典型的：生产者、消费者模型。生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，这样就实现了生产者和消费者的解耦。

开发中消息队列通常有如下应用场景：

1、任务异步处理：

高并发环境下，由于来不及同步处理，请求往往会发生堵塞，比如说，大量的insert，update之类的请求同时到达MySQL，直接导致无数的行锁表锁，甚至最后请求会堆积过多，从而触发too many connections错误。通过使用消息队列，我们可以异步处理请求，从而缓解系统的压力。将不需要同步处理的并且耗时长的操作由消息队列通知消息接收方进行异步处理。减少了应用程序的响应时间。

2、应用程序解耦合：

MQ相当于一个中介，生产方通过MQ与消费方交互，它将应用程序进行解耦合。

AMQP和JMS

MQ是消息通信的模型，并发具体实现。现在实现MQ的有两种主流方式：AMQP、JMS。

两者间的区别和联系：

  • JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式
  • JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
  • JMS规定了两种消息模型；而AMQP的消息模型更加丰富

常见MQ产品

ActiveMQ：基于JMS

RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好

RocketMQ：基于JMS，阿里巴巴产品，目前交由Apache基金会

Kafka：分布式消息系统，高吞吐量 

RabbitMQ快速入门

RabbitMQ是由erlang语言开发，基于AMQP（Advanced Message Queue 高级消息队列协议）协议实现的消息队列，它是一种应用程序之间的通信方法，消息队列在分布式系统开发中应用非常广泛。RabbitMQ官方地址：http://www.rabbitmq.com

下载与安装

RabbitMQ由Erlang语言开发，需要安装与RabbitMQ版本对应的Erlang语言环境，具体的就不解释了，自行搜索教程。RabbitMQ官网下载地址：http://www.rabbitmq.com/download.html

RabbitMQ的工作原理

下图是RabbitMQ的基本结构：

 组成部分说明：

  • Broker：消息队列服务进程，此进程包括两个部分：Exchange和Queue
  • Exchange：消息队列交换机，按一定的规则将消息路由转发到某个队列，对消息进行过虑。
  • Queue：消息队列，存储消息的队列，消息到达队列并转发给指定的
  • Producer：消息生产者，即生产方客户端，生产方客户端将消息发送
  • Consumer：消息消费者，即消费方客户端，接收MQ转发的消息。

生产者发送消息流程：

1、生产者和Broker建立TCP连接。

2、生产者和Broker建立通道。

3、生产者通过通道消息发送给Broker，由Exchange将消息进行转发。

4、Exchange将消息转发到指定的Queue（队列）

消费者接收消息流程：

1、消费者和Broker建立TCP连接

2、消费者和Broker建立通道

3、消费者监听指定的Queue（队列）

4、当有消息到达Queue时Broker默认将消息推送给消费者。

5、消费者接收到消息。

6、ack回复

六种消息模型

①基本消息模型：


在上图的模型中，有以下概念：

  • P：生产者，也就是要发送消息的程序
  • C：消费者：消息的接受者，会一直等待消息到来。
  • queue：消息队列，图中红色部分。可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。 

 生产者

新建一个maven工程，添加amqp-client依赖

  1. <dependency>
  2.             <groupId>com.rabbitmq</groupId>
  3.             <artifactId>amqp-client</artifactId>
  4.             <version>5.7.1</version>
  5. </dependency>


 连接工具类：

  1. public class ConnectionUtil {
  2.     /**
  3.      * 建立与RabbitMQ的连接
  4.      * @return
  5.      * @throws Exception
  6.      */
  7.     public static Connection getConnection() throws Exception {
  8.         //定义连接工厂
  9.         ConnectionFactory factory = new ConnectionFactory();
  10.         //设置服务地址
  11.         factory.setHost("192.168.1.103");
  12.         //端口
  13.         factory.setPort(5672);
  14.         //设置账号信息，用户名、密码、vhost
  15.         factory.setVirtualHost("/kavito");//设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq
  16.         factory.setUsername("kavito");
  17.         factory.setPassword("123456");
  18.         // 通过工厂获取连接
  19.         Connection connection = factory.newConnection();
  20.         return connection;
  21.     }
  22. }

生产者发送消息：

  1. public class Send {
  2.  
  3.     private final static String QUEUE_NAME = "simple_queue";
  4.  
  5.     public static void main(String[] argv) throws Exception {
  6.         // 1、获取到连接
  7.         Connection connection = ConnectionUtil.getConnection();
  8.         // 2、从连接中创建通道，使用通道才能完成消息相关的操作
  9.         Channel channel = connection.createChannel();
  10.         // 3、声明（创建）队列
  11.         //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
  12.         /**
  13.          * 参数明细
  14.          * 1、queue 队列名称
  15.          * 2、durable 是否持久化，如果持久化，mq重启后队列还在
  16.          * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
  17.          * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
  18.          * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
  19.          */
  20.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  21.         // 4、消息内容
  22.         String message = "Hello World!";
  23.         // 向指定的队列中发送消息
  24.         //参数：String exchange, String routingKey, BasicProperties props, byte[] body
  25.         /**
  26.          * 参数明细：
  27.          * 1、exchange，交换机，如果不指定将使用mq的默认交换机（设置为""）
  28.          * 2、routingKey，路由key，交换机根据路由key来将消息转发到指定的队列，如果使用默认交换机，routingKey设置为队列的名称
  29.          * 3、props，消息的属性
  30.          * 4、body，消息内容
  31.          */
  32.         channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
  33.         System.out.println(" [x] Sent '" + message + "'");
  34.         
  35.         //关闭通道和连接(资源关闭最好用try-catch-finally语句处理)
  36.         channel.close();
  37.         connection.close();
  38.     }
  39. }

 控制台：

web管理页面：服务器地址/端口号 （本地：127.0.0.1:15672，默认用户及密码：guest guest）

点击队列名称，进入详情页，可以查看消息：

消费者接收消息

  1. public class Recv {
  2.     private final static String QUEUE_NAME = "simple_queue";
  3.  
  4.     public static void main(String[] argv) throws Exception {
  5.         // 获取到连接
  6.         Connection connection = ConnectionUtil.getConnection();
  7.         //创建会话通道,生产者和mq服务所有通信都在channel通道中完成
  8.         Channel channel = connection.createChannel();
  9.         // 声明队列
  10.         //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
  11.         /**
  12.          * 参数明细
  13.          * 1、queue 队列名称
  14.          * 2、durable 是否持久化，如果持久化，mq重启后队列还在
  15.          * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
  16.          * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
  17.          * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
  18.          */
  19.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  20.         //实现消费方法
  21.         DefaultConsumer consumer = new DefaultConsumer(channel){
  22.             // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
  23.             /**
  24.              * 当接收到消息后此方法将被调用
  25.              * @param consumerTag  消费者标签，用来标识消费者的，在监听队列时设置channel.basicConsume
  26.              * @param envelope 信封，通过envelope
  27.              * @param properties 消息属性
  28.              * @param body 消息内容
  29.              * @throws IOException
  30.              */
  31.             @Override
  32.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  33.                 //交换机
  34.                 String exchange = envelope.getExchange();
  35.                 //消息id，mq在channel中用来标识消息的id，可用于确认消息已接收
  36.                 long deliveryTag = envelope.getDeliveryTag();
  37.                 // body 即消息体
  38.                 String msg = new String(body,"utf-8");
  39.                 System.out.println(" [x] received : " + msg + "!");
  40.             }
  41.         };
  42.         
  43.         // 监听队列，第二个参数：是否自动进行消息确认。
  44.         //参数：String queue, boolean autoAck, Consumer callback
  45.         /**
  46.          * 参数明细：
  47.          * 1、queue 队列名称
  48.          * 2、autoAck 自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq，如果设置为false要通过编程实现回复
  49.          * 3、callback，消费方法，当消费者接收到消息要执行的方法
  50.          */
  51.         channel.basicConsume(QUEUE_NAME, true, consumer);
  52.     }
  53. }

控制台打印：

再看看队列的消息,已经被消费了

我们发现，消费者已经获取了消息，但是程序没有停止，一直在监听队列中是否有新的消息。一旦有新的消息进入队列，就会立即打印.

消息确认机制（ACK）

通过刚才的案例可以看出，消息一旦被消费者接收，队列中的消息就会被删除。

那么问题来了：RabbitMQ怎么知道消息被接收了呢？

如果消费者领取消息后，还没执行操作就挂掉了呢？或者抛出了异常？消息消费失败，但是RabbitMQ无从得知，这样消息就丢失了！

因此，RabbitMQ有一个ACK机制。当消费者获取消息后，会向RabbitMQ发送回执ACK，告知消息已经被接收。不过这种回执ACK分两种情况：

  • 自动ACK：消息一旦被接收，消费者自动发送ACK
  • 手动ACK：消息接收后，不会发送ACK，需要手动调用

大家觉得哪种更好呢？

这需要看消息的重要性：

  • 如果消息不太重要，丢失也没有影响，那么自动ACK会比较方便
  • 如果消息非常重要，不容丢失。那么最好在消费完成后手动ACK，否则接收消息后就自动ACK，RabbitMQ就会把消息从队列中删除。如果此时消费者宕机，那么消息就丢失了。

我们之前的测试都是自动ACK的，如果要手动ACK，需要改动我们的代码：

  1. public class Recv2 {
  2.     private final static String QUEUE_NAME = "simple_queue";
  3.  
  4.     public static void main(String[] argv) throws Exception {
  5.         // 获取到连接
  6.         Connection connection = ConnectionUtil.getConnection();
  7.         // 创建通道
  8.         final Channel channel = connection.createChannel();
  9.         // 声明队列
  10.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  11.         // 定义队列的消费者
  12.         DefaultConsumer consumer = new DefaultConsumer(channel) {
  13.             // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
  14.             @Override
  15.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  16.                 // body 即消息体
  17.                 String msg = new String(body);
  18.                 System.out.println(" [x] received : " + msg + "!");
  19.                 // 手动进行ACK
  20.                 /*
  21.                  *  void basicAck(long deliveryTag, boolean multiple) throws IOException;
  22.                  *  deliveryTag:用来标识消息的id
  23.                  *  multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息。
  24.                  */
  25.                 
  26.                 channel.basicAck(envelope.getDeliveryTag(), false);
  27.             }
  28.         };
  29.         // 监听队列，第二个参数false，手动进行ACK
  30.         channel.basicConsume(QUEUE_NAME, false, consumer);
  31.     }
  32. }

 最后一行代码设置第二个参数为false

channel.basicConsume(QUEUE_NAME, false, consumer);

自动ACK存在的问题

修改消费者，添加异常，如下：

生产者不做任何修改，直接运行，消息发送成功：  

 运行消费者，程序抛出异常：

管理界面：

 消费者抛出异常，但是消息依然被消费，实际上我们还没获取到消息。

演示手动ACK

重新运行生产者发送消息：

同样，在手动进行ack前抛出异常，运行Recv2

 再看看管理界面：

  消息没有被消费掉！

还有另外一种情况：修改消费者Recv2，把监听队列第二个参数自动改成手动,（去掉之前制造的异常） ，并且消费方法中没手动进行ACK

 生产者代码不变，再次运行：

运行消费者 ：

但是，查看管理界面，发现：

停掉消费者的程序，发现：

 这是因为虽然我们设置了手动ACK，但是代码中并没有进行消息确认！所以消息并未被真正消费掉。当我们关掉这个消费者，消息的状态再次变为Ready。

正确的做法是：

我们要在监听队列时设置第二个参数为false,代码中手动进行ACK

 再次运行消费者，查看web管理页面：

消费者消费成功！  

生产者避免数据丢失：https://www.cnblogs.com/vipstone/p/9350075.html

②work消息模型

工作队列或者竞争消费者模式

work queues与入门程序相比，多了一个消费端，两个消费端共同消费同一个队列中的消息，但是一个消息只能被一个消费者获取。

这个消息模型在Web应用程序中特别有用，可以处理短的HTTP请求窗口中无法处理复杂的任务。

接下来我们来模拟这个流程：

P：生产者：任务的发布者

C1：消费者1：领取任务并且完成任务，假设完成速度较慢（模拟耗时）

C2：消费者2：领取任务并且完成任务，假设完成速度较快

生产者

生产者循环发送50条消息

  1. public class Send {
  2.     private final static String QUEUE_NAME = "test_work_queue";
  3.  
  4.     public static void main(String[] argv) throws Exception {
  5.         // 获取到连接
  6.         Connection connection = ConnectionUtil.getConnection();
  7.         // 获取通道
  8.         Channel channel = connection.createChannel();
  9.         // 声明队列
  10.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  11.         // 循环发布任务
  12.         for (int i = 0; i < 50; i++) {
  13.             // 消息内容
  14.             String message = "task .. " + i;
  15.             channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
  16.             System.out.println(" [x] Sent '" + message + "'");
  17.  
  18.             Thread.sleep(i * 2);
  19.         }
  20.         // 关闭通道和连接
  21.         channel.close();
  22.         connection.close();
  23.     }
  24. }

消费者1

  1. public class Recv {
  2.     private final static String QUEUE_NAME = "test_work_queue";
  3.  
  4.     public static void main(String[] argv) throws Exception {
  5.         // 获取到连接
  6.         Connection connection = ConnectionUtil.getConnection();
  7.         //创建会话通道,生产者和mq服务所有通信都在channel通道中完成
  8.         Channel channel = connection.createChannel();
  9.         // 声明队列
  10.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  11.         //实现消费方法
  12.         DefaultConsumer consumer = new DefaultConsumer(channel){
  13.             // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
  14.             @Override
  15.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  16.                 // body 即消息体
  17.                 String msg = new String(body,"utf-8");
  18.                 System.out.println(" [消费者1] received : " + msg + "!");
  19.                 //模拟任务耗时1s
  20.                 try { TimeUnit.SECONDS.sleep(1); } catch (Exception e) { e.printStackTrace(); }
  21.             }
  22.         };
  23.         // 监听队列，第二个参数：是否自动进行消息确认。
  24.         channel.basicConsume(QUEUE_NAME, true, consumer);
  25.     }
  26. }

消费者2
代码不贴了，与消费者1基本类似，只是消费者2没有设置消费耗时时间。

接下来，两个消费者一同启动，然后发送50条消息：

 

可以发现，两个消费者各自消费了不同25条消息，这就实现了任务的分发。   

能者多劳

刚才的实现有问题吗？

  • 消费者1比消费者2的效率要低，一次任务的耗时较长
  • 然而两人最终消费的消息数量是一样的
  • 消费者2大量时间处于空闲状态，消费者1一直忙碌

现在的状态属于是把任务平均分配，正确的做法应该是消费越快的人，消费的越多。

怎么实现呢？

通过 BasicQos 方法设置prefetchCount = 1。这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理1个Message。换句话说，在接收到该Consumer的ack前，他它不会将新的Message分发给它。相反，它会将其分派给不是仍然忙碌的下一个Consumer。

值得注意的是：prefetchCount在手动ack的情况下才生效，自动ack不生效。

再次测试： 

 

订阅模型分类

说明下：

1、一个生产者多个消费者
2、每个消费者都有一个自己的队列
3、生产者没有将消息直接发送给队列，而是发送给exchange(交换机、转发器)
4、每个队列都需要绑定到交换机上
5、生产者发送的消息，经过交换机到达队列，实现一个消息被多个消费者消费
例子：注册->发邮件、发短信

X（Exchanges）：交换机一方面：接收生产者发送的消息。另一方面：知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。

Exchange类型有以下几种：

Fanout：广播，将消息交给所有绑定到交换机的队列

Direct：定向，把消息交给符合指定routing key 的队列

Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列

Header：header模式与routing不同的地方在于，header模式取消routingkey，使用header中的 key/value（键值对）匹配队列。

Header模式不展开了，感兴趣可以参考这篇文章https://blog.csdn.net/zhu_tianwei/article/details/40923131

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

③Publish/subscribe（交换机类型：Fanout，也称为广播 ）

Publish/subscribe模型示意图 ：

生产者

和前面两种模式不同：

  • 1） 声明Exchange，不再声明Queue
  • 2） 发送消息到Exchange，不再发送到Queue
  1. public class Send {
  2.  
  3.     private final static String EXCHANGE_NAME = "test_fanout_exchange";
  4.  
  5.     public static void main(String[] argv) throws Exception {
  6.         // 获取到连接
  7.         Connection connection = ConnectionUtil.getConnection();
  8.         // 获取通道
  9.         Channel channel = connection.createChannel();
  10.         // 声明exchange，指定类型为fanout
  11.         channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
  12.         
  13.         // 消息内容
  14.         String message = "注册成功！！";
  15.         // 发布消息到Exchange
  16.         channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
  17.         System.out.println(" [生产者] Sent '" + message + "'");
  18.  
  19.         channel.close();
  20.         connection.close();
  21.     }
  22. }

消费者1 （注册成功发给短信服务）

  1. public class Recv {
  2.     private final static String QUEUE_NAME = "fanout_exchange_queue_sms";//短信队列
  3.  
  4.     private final static String EXCHANGE_NAME = "test_fanout_exchange";
  5.  
  6.     public static void main(String[] argv) throws Exception {
  7.         // 获取到连接
  8.         Connection connection = ConnectionUtil.getConnection();
  9.         // 获取通道
  10.         Channel channel = connection.createChannel();
  11.         // 声明队列
  12.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  13.  
  14.         // 绑定队列到交换机
  15.         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
  16.  
  17.         // 定义队列的消费者
  18.         DefaultConsumer consumer = new DefaultConsumer(channel) {
  19.             // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
  20.             @Override
  21.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
  22.                                        byte[] body) throws IOException {
  23.                 // body 即消息体
  24.                 String msg = new String(body);
  25.                 System.out.println(" [短信服务] received : " + msg + "!");
  26.             }
  27.         };
  28.         // 监听队列，自动返回完成
  29.         channel.basicConsume(QUEUE_NAME, true, consumer);
  30.     }
  31. }

消费者2（注册成功发给邮件服务）

  1. public class Recv2 {
  2.     private final static String QUEUE_NAME = "fanout_exchange_queue_email";//邮件队列
  3.  
  4.     private final static String EXCHANGE_NAME = "test_fanout_exchange";
  5.  
  6.     public static void main(String[] argv) throws Exception {
  7.         // 获取到连接
  8.         Connection connection = ConnectionUtil.getConnection();
  9.         // 获取通道
  10.         Channel channel = connection.createChannel();
  11.         // 声明队列
  12.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  13.  
  14.         // 绑定队列到交换机
  15.         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
  16.         
  17.         // 定义队列的消费者
  18.         DefaultConsumer consumer = new DefaultConsumer(channel) {
  19.             // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
  20.             @Override
  21.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
  22.                                        byte[] body) throws IOException {
  23.                 // body 即消息体
  24.                 String msg = new String(body);
  25.                 System.out.println(" [邮件服务] received : " + msg + "!");
  26.             }
  27.         };
  28.         // 监听队列，自动返回完成
  29.         channel.basicConsume(QUEUE_NAME, true, consumer);
  30.     }
  31. }

 我们运行两个消费者，然后发送1条消息：

 

思考：

1、publish/subscribe与work queues有什么区别。

区别：

1）work queues不用定义交换机，而publish/subscribe需要定义交换机。

2）publish/subscribe的生产方是面向交换机发送消息，work queues的生产方是面向队列发送消息(底层使用默认交换机)。

3）publish/subscribe需要设置队列和交换机的绑定，work queues不需要设置，实际上work queues会将队列绑定到默认的交换机 。

相同点：

所以两者实现的发布/订阅的效果是一样的，多个消费端监听同一个队列不会重复消费消息。

2、实际工作用 publish/subscribe还是work queues。

建议使用 publish/subscribe，发布订阅模式比工作队列模式更强大（也可以做到同一队列竞争），并且发布订阅模式可以指定自己专用的交换机。

④Routing 路由模型（交换机类型：direct）

Routing模型示意图：

 

P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。

X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列

C1：消费者，其所在队列指定了需要routing key 为 error 的消息

C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

接下来看代码：

生产者

  1. public class Send {
  2.     private final static String EXCHANGE_NAME = "test_direct_exchange";
  3.  
  4.     public static void main(String[] argv) throws Exception {
  5.         // 获取到连接
  6.         Connection connection = ConnectionUtil.getConnection();
  7.         // 获取通道
  8.         Channel channel = connection.createChannel();
  9.         // 声明exchange，指定类型为direct
  10.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
  11.         // 消息内容，
  12.         String message = "注册成功！请短信回复[T]退订";
  13.         // 发送消息，并且指定routing key 为：sms，只有短信服务能接收到消息
  14.         channel.basicPublish(EXCHANGE_NAME, "sms", null, message.getBytes());
  15.         System.out.println(" [x] Sent '" + message + "'");
  16.  
  17.         channel.close();
  18.         connection.close();
  19.     }
  20. }

消费者1

  1. public class Recv {
  2.     private final static String QUEUE_NAME = "direct_exchange_queue_sms";//短信队列
  3.     private final static String EXCHANGE_NAME = "test_direct_exchange";
  4.  
  5.     public static void main(String[] argv) throws Exception {
  6.         // 获取到连接
  7.         Connection connection = ConnectionUtil.getConnection();
  8.         // 获取通道
  9.         Channel channel = connection.createChannel();
  10.         // 声明队列
  11.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  12.         
  13.         // 绑定队列到交换机，同时指定需要订阅的routing key。可以指定多个
  14.         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "sms");//指定接收发送方指定routing key为sms的消息
  15.         //channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "email");
  16.  
  17.         // 定义队列的消费者
  18.         DefaultConsumer consumer = new DefaultConsumer(channel) {
  19.             // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
  20.             @Override
  21.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
  22.                                        byte[] body) throws IOException {
  23.                 // body 即消息体
  24.                 String msg = new String(body);
  25.                 System.out.println(" [短信服务] received : " + msg + "!");
  26.             }
  27.         };
  28.         // 监听队列，自动ACK
  29.         channel.basicConsume(QUEUE_NAME, true, consumer);
  30.     }
  31. }

消费者2 

  1. public class Recv2 {
  2.     private final static String QUEUE_NAME = "direct_exchange_queue_email";//邮件队列
  3.     private final static String EXCHANGE_NAME = "test_direct_exchange";
  4.  
  5.     public static void main(String[] argv) throws Exception {
  6.         // 获取到连接
  7.         Connection connection = ConnectionUtil.getConnection();
  8.         // 获取通道
  9.         Channel channel = connection.createChannel();
  10.         // 声明队列
  11.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  12.         
  13.         // 绑定队列到交换机，同时指定需要订阅的routing key。可以指定多个
  14.         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "email");//指定接收发送方指定routing key为email的消息
  15.  
  16.         // 定义队列的消费者
  17.         DefaultConsumer consumer = new DefaultConsumer(channel) {
  18.             // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
  19.             @Override
  20.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
  21.                                        byte[] body) throws IOException {
  22.                 // body 即消息体
  23.                 String msg = new String(body);
  24.                 System.out.println(" [邮件服务] received : " + msg + "!");
  25.             }
  26.         };
  27.         // 监听队列，自动ACK
  28.         channel.basicConsume(QUEUE_NAME, true, consumer);
  29.     }
  30. }

我们发送sms的RoutingKey，发现结果：只有指定短信的消费者1收到消息了

⑤Topics 通配符模式（交换机类型：topics）

Topics模型示意图：

每个消费者监听自己的队列，并且设置带统配符的routingkey,生产者将消息发给broker，由交换机根据routingkey来转发消息到指定的队列。

Routingkey一般都是有一个或者多个单词组成，多个单词之间以“.”分割，例如：inform.sms

通配符规则：

#：匹配一个或多个词

*：匹配不多不少恰好1个词

举例：

audit.#：能够匹配audit.irs.corporate 或者 audit.irs

audit.*：只能匹配audit.irs

从示意图可知，我们将发送所有描述动物的消息。消息将使用由三个字（两个点）组成的Routing key发送。路由关键字中的第一个单词将描述速度，第二个颜色和第三个种类：“<speed>.<color>.<species>”。

我们创建了三个绑定：Q1绑定了“*.orange.*”，Q2绑定了“.*.*.rabbit”和“lazy.＃”。

Q1匹配所有的橙色动物。

Q2匹配关于兔子以及懒惰动物的消息。

 下面做个小练习，假如生产者发送如下消息，会进入哪个队列：

quick.orange.rabbit       Q1 Q2   routingKey="quick.orange.rabbit"的消息会同时路由到Q1与Q2

lazy.orange.elephant    Q1 Q2

quick.orange.fox           Q1

lazy.pink.rabbit              Q2  (值得注意的是，虽然这个routingKey与Q2的两个bindingKey都匹配，但是只会投递Q2一次)

quick.brown.fox            不匹配任意队列，被丢弃

quick.orange.male.rabbit   不匹配任意队列，被丢弃

orange         不匹配任意队列，被丢弃

下面我们以指定Routing key="quick.orange.rabbit"为例，验证上面的答案

生产者

  1. public class Send {
  2.     private final static String EXCHANGE_NAME = "test_topic_exchange";
  3.  
  4.     public static void main(String[] argv) throws Exception {
  5.         // 获取到连接
  6.         Connection connection = ConnectionUtil.getConnection();
  7.         // 获取通道
  8.         Channel channel = connection.createChannel();
  9.         // 声明exchange，指定类型为topic
  10.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
  11.         // 消息内容
  12.         String message = "这是一只行动迅速的橙色的兔子";
  13.         // 发送消息，并且指定routing key为：quick.orange.rabbit
  14.         channel.basicPublish(EXCHANGE_NAME, "quick.orange.rabbit", null, message.getBytes());
  15.         System.out.println(" [动物描述：] Sent '" + message + "'");
  16.  
  17.         channel.close();
  18.         connection.close();
  19.     }
  20. }

消费者1

  1. public class Recv {
  2.     private final static String QUEUE_NAME = "topic_exchange_queue_Q1";
  3.     private final static String EXCHANGE_NAME = "test_topic_exchange";
  4.  
  5.     public static void main(String[] argv) throws Exception {
  6.         // 获取到连接
  7.         Connection connection = ConnectionUtil.getConnection();
  8.         // 获取通道
  9.         Channel channel = connection.createChannel();
  10.         // 声明队列
  11.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  12.         
  13.         // 绑定队列到交换机，同时指定需要订阅的routing key。订阅所有的橙色动物
  14.         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.orange.*");
  15.  
  16.         // 定义队列的消费者
  17.         DefaultConsumer consumer = new DefaultConsumer(channel) {
  18.             // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
  19.             @Override
  20.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
  21.                                        byte[] body) throws IOException {
  22.                 // body 即消息体
  23.                 String msg = new String(body);
  24.                 System.out.println(" [消费者1] received : " + msg + "!");
  25.             }
  26.         };
  27.         // 监听队列，自动ACK
  28.         channel.basicConsume(QUEUE_NAME, true, consumer);
  29.     }
  30. }

 消费者2

  1. public class Recv2 {
  2.     private final static String QUEUE_NAME = "topic_exchange_queue_Q2";
  3.     private final static String EXCHANGE_NAME = "test_topic_exchange";
  4.  
  5.     public static void main(String[] argv) throws Exception {
  6.         // 获取到连接
  7.         Connection connection = ConnectionUtil.getConnection();
  8.         // 获取通道
  9.         Channel channel = connection.createChannel();
  10.         // 声明队列
  11.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  12.         
  13.         // 绑定队列到交换机，同时指定需要订阅的routing key。订阅关于兔子以及懒惰动物的消息
  14.         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.*.rabbit");
  15.         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "lazy.＃");
  16.  
  17.         // 定义队列的消费者
  18.         DefaultConsumer consumer = new DefaultConsumer(channel) {
  19.             // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
  20.             @Override
  21.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
  22.                                        byte[] body) throws IOException {
  23.                 // body 即消息体
  24.                 String msg = new String(body);
  25.                 System.out.println(" [消费者2] received : " + msg + "!");
  26.             }
  27.         };
  28.         // 监听队列，自动ACK
  29.         channel.basicConsume(QUEUE_NAME, true, consumer);
  30.     }
  31. }

结果C1、C2是都接收到消息了：

⑥RPC

RPC模型示意图：

 

基本概念：

Callback queue 回调队列，客户端向服务器发送请求，服务器端处理请求后，将其处理结果保存在一个存储体中。而客户端为了获得处理结果，那么客户在向服务器发送请求时，同时发送一个回调队列地址reply_to。

Correlation id 关联标识，客户端可能会发送多个请求给服务器，当服务器处理完后，客户端无法辨别在回调队列中的响应具体和那个请求时对应的。为了处理这种情况，客户端在发送每个请求时，同时会附带一个独有correlation_id属性，这样客户端在回调队列中根据correlation_id字段的值就可以分辨此响应属于哪个请求。

流程说明：

  • 当客户端启动的时候，它创建一个匿名独享的回调队列。
  • 在 RPC 请求中，客户端发送带有两个属性的消息：一个是设置回调队列的 reply_to 属性，另一个是设置唯一值的 correlation_id 属性。
  • 将请求发送到一个 rpc_queue 队列中。
  • 服务器等待请求发送到这个队列中来。当请求出现的时候，它执行他的工作并且将带有执行结果的消息发送给 reply_to 字段指定的队列。
  • 客户端等待回调队列里的数据。当有消息出现的时候，它会检查 correlation_id 属性。如果此属性的值与请求匹配，将它返回给应用。

使用RabbitMQ搭建一个RPC系统：一个客户端和一个可以升级(扩展)的RPC服务器。为了模拟一个耗时任务，我们将创建一个返回斐波那契数列的虚拟的RPC服务。

客户端

在客户端定义一个RPCClient类,并定义一个call()方法,这个方法发送一个RPC请求,并等待接收响应结果

  1. RPCClient client = new RPCClient();
  2. String result = client.call("4");
  3. System.out.println( "第四个斐波那契数是: " + result);

回调队列 Callback Queue

使用RabbitMQ去实现RPC很容易。一个客户端发送请求信息，并得到一个服务器端回复的响应信息。为了得到响应信息，我们需要在请求的时候发送一个“回调”队列地址。我们可以使用默认队列。下面是示例代码：

  1. //定义回调队列,
  2. //自动生成对列名,非持久,独占,自动删除
  3. callbackQueueName = ch.queueDeclare().getQueue();
  4.  
  5. //用来设置回调队列的参数对象
  6. BasicProperties props = new BasicProperties
  7.                             .Builder()
  8.                             .replyTo(callbackQueueName)
  9.                             .build();
  10. //发送调用消息
  11. ch.basicPublish("", "rpc_queue", props, message.getBytes());

消息属性 Message Properties

AMQP 0-9-1协议定义了消息的14个属性。大部分属性很少使用，下面是比较常用的4个:

  • deliveryMode:将消息标记为持久化(值为2)或非持久化(任何其他值)。
  • contentType:用于描述mime类型。例如，对于经常使用的JSON格式，将此属性设置为:application/json。
  • replyTo:通常用于指定回调队列。
  • correlationId:将RPC响应与请求关联起来非常有用。

关联id (correlationId)：

在上面的代码中，我们会为每个RPC请求创建一个回调队列。 这是非常低效的，这里还有一个更好的方法:让我们为每个客户端创建一个回调队列。

这就提出了一个新的问题，在队列中得到一个响应时，我们不清楚这个响应所对应的是哪一条请求。这时候就需要使用关联id（correlationId）。我们将为每一条请求设置唯一的的id值。稍后，当我们在回调队列里收到一条消息的时候，我们将查看它的id属性，这样我们就可以匹配对应的请求和响应。如果我们发现了一个未知的id值，我们可以安全的丢弃这条消息，因为它不属于我们的请求。

代码

服务器端:

  1. package rabbitmq.rpc;
  2.  
  3. import java.io.IOException;
  4. import java.util.Random;
  5. import java.util.Scanner;
  6.  
  7. import com.rabbitmq.client.AMQP;
  8. import com.rabbitmq.client.BuiltinExchangeType;
  9. import com.rabbitmq.client.CancelCallback;
  10. import com.rabbitmq.client.Channel;
  11. import com.rabbitmq.client.Connection;
  12. import com.rabbitmq.client.ConnectionFactory;
  13. import com.rabbitmq.client.DeliverCallback;
  14. import com.rabbitmq.client.Delivery;
  15. import com.rabbitmq.client.AMQP.BasicProperties;
  16.  
  17. public class RPCServer {
  18. 	public static void main(String[] args) throws Exception {
  19. 		ConnectionFactory f = new ConnectionFactory();
  20. 		f.setHost("192.168.64.140");
  21. 		f.setPort(5672);
  22. 		f.setUsername("admin");
  23. 		f.setPassword("admin");
  24. 		
  25. 		Connection c = f.newConnection();
  26. 		Channel ch = c.createChannel();
  27. 		/*
  28. 		 * 定义队列 rpc_queue, 将从它接收请求信息
  29. 		 * 
  30. 		 * 参数:
  31. 		 * 1. queue, 对列名
  32. 		 * 2. durable, 持久化
  33. 		 * 3. exclusive, 排他
  34. 		 * 4. autoDelete, 自动删除
  35. 		 * 5. arguments, 其他参数属性
  36. 		 */
  37. 		ch.queueDeclare("rpc_queue",false,false,false,null);
  38. 		ch.queuePurge("rpc_queue");//清除队列中的内容
  39. 		
  40. 		ch.basicQos(1);//一次只接收一条消息
  41. 		
  42. 		
  43. 		//收到请求消息后的回调对象
  44. 		DeliverCallback deliverCallback = new DeliverCallback() {
  45. 			@Override
  46. 			public void handle(String consumerTag, Delivery message) throws IOException {
  47. 				//处理收到的数据(要求第几个斐波那契数)
  48. 				String msg = new String(message.getBody(), "UTF-8");
  49. 				int n = Integer.parseInt(msg);
  50. 				//求出第n个斐波那契数
  51. 				int r = fbnq(n);
  52. 				String response = String.valueOf(r);
  53. 				
  54. 				//设置发回响应的id, 与请求id一致, 这样客户端可以把该响应与它的请求进行对应
  55. 				BasicProperties replyProps = new BasicProperties.Builder()
  56. 						.correlationId(message.getProperties().getCorrelationId())
  57. 						.build();
  58. 				/*
  59. 				 * 发送响应消息
  60. 				 * 1. 默认交换机
  61. 				 * 2. 由客户端指定的,用来传递响应消息的队列名
  62. 				 * 3. 参数(关联id)
  63. 				 * 4. 发回的响应消息
  64. 				 */
  65. 				ch.basicPublish("",message.getProperties().getReplyTo(), replyProps, response.getBytes("UTF-8"));
  66. 				//发送确认消息
  67. 				ch.basicAck(message.getEnvelope().getDeliveryTag(), false);
  68. 			}
  69. 		};
  70. 		
  71. 		//
  72. 		CancelCallback cancelCallback = new CancelCallback() {
  73. 			@Override
  74. 			public void handle(String consumerTag) throws IOException {
  75. 			}
  76. 		};
  77. 		
  78. 		//消费者开始接收消息, 等待从 rpc_queue接收请求消息, 不自动确认
  79. 		ch.basicConsume("rpc_queue", false, deliverCallback, cancelCallback);
  80. 	}
  81.  
  82. 	protected static int fbnq(int n) {
  83. 		if(n == 1 || n == 2) return 1;
  84. 		
  85. 		return fbnq(n-1)+fbnq(n-2);
  86. 	}
  87. }

客户端:

  1. package rabbitmq.rpc;
  2.  
  3. import java.io.IOException;
  4. import java.util.Scanner;
  5. import java.util.UUID;
  6. import java.util.concurrent.ArrayBlockingQueue;
  7. import java.util.concurrent.BlockingQueue;
  8.  
  9. import com.rabbitmq.client.BuiltinExchangeType;
  10. import com.rabbitmq.client.CancelCallback;
  11. import com.rabbitmq.client.Channel;
  12. import com.rabbitmq.client.Connection;
  13. import com.rabbitmq.client.ConnectionFactory;
  14. import com.rabbitmq.client.DeliverCallback;
  15. import com.rabbitmq.client.Delivery;
  16. import com.rabbitmq.client.AMQP.BasicProperties;
  17.  
  18. public class RPCClient {
  19. 	Connection con;
  20. 	Channel ch;
  21. 	
  22. 	public RPCClient() throws Exception {
  23. 		ConnectionFactory f = new ConnectionFactory();
  24. 		f.setHost("192.168.64.140");
  25. 		f.setUsername("admin");
  26. 		f.setPassword("admin");
  27. 		con = f.newConnection();
  28. 		ch = con.createChannel();
  29. 	}
  30. 	
  31. 	public String call(String msg) throws Exception {
  32. 		//自动生成对列名,非持久,独占,自动删除
  33. 		String replyQueueName = ch.queueDeclare().getQueue();
  34. 		//生成关联id
  35. 		String corrId = UUID.randomUUID().toString();
  36. 		
  37. 		//设置两个参数:
  38. 		//1. 请求和响应的关联id
  39. 		//2. 传递响应数据的queue
  40. 		BasicProperties props = new BasicProperties.Builder()
  41. 				.correlationId(corrId)
  42. 				.replyTo(replyQueueName)
  43. 				.build();
  44. 		//向 rpc_queue 队列发送请求数据, 请求第n个斐波那契数
  45. 		ch.basicPublish("", "rpc_queue", props, msg.getBytes("UTF-8"));
  46. 		
  47. 		//用来保存结果的阻塞集合,取数据时,没有数据会暂停等待
  48. 		BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);
  49. 		
  50. 		//接收响应数据的回调对象
  51. 		DeliverCallback deliverCallback = new DeliverCallback() {
  52. 			@Override
  53. 			public void handle(String consumerTag, Delivery message) throws IOException {
  54. 				//如果响应消息的关联id,与请求的关联id相同,我们来处理这个响应数据
  55. 				if (message.getProperties().getCorrelationId().contentEquals(corrId)) {
  56. 					//把收到的响应数据,放入阻塞集合
  57. 					response.offer(new String(message.getBody(), "UTF-8"));
  58. 				}
  59. 			}
  60. 		};
  61.  
  62. 		CancelCallback cancelCallback = new CancelCallback() {
  63. 			@Override
  64. 			public void handle(String consumerTag) throws IOException {
  65. 			}
  66. 		};
  67. 		
  68. 		//开始从队列接收响应数据
  69. 		ch.basicConsume(replyQueueName, true, deliverCallback, cancelCallback);
  70. 		//返回保存在集合中的响应数据
  71. 		return response.take();
  72. 	}
  73. 	
  74. 	public static void main(String[] args) throws Exception {
  75. 		RPCClient client = new RPCClient();
  76. 		while (true) {
  77. 			System.out.print("求第几个斐波那契数:");
  78. 			int n = new Scanner(System.in).nextInt();
  79. 			String r = client.call(""+n);
  80. 			System.out.println(r);
  81. 		}
  82. 	}
  83. }

分享两道面试题：

1.避免消息堆积？

1） 采用workqueue，多个消费者监听同一队列。

2）接收到消息以后，而是通过线程池，异步消费。

2.如何避免消息丢失？

1） 消费者的ACK机制。可以防止消费者丢失消息。

但是，如果在消费者消费之前，MQ就宕机了，消息就没了？

2）可以将消息进行持久化。要将消息持久化，前提是：队列、Exchange都持久化

交换机持久化


队列持久化

消息持久化

Spring整合RibbitMQ

下面还是模拟注册服务当用户注册成功后，向短信和邮件服务推送消息的场景

搭建SpringBoot环境

创建两个工程 mq-rabbitmq-producer和mq-rabbitmq-consumer，分别配置1、2、3（第三步本例消费者用注解形式，可以不用配）

1、添加AMQP的启动器：

  1. <dependency>
  2.     <groupId>org.springframework.boot</groupId>
  3.     <artifactId>spring-boot-starter-amqp</artifactId>
  4. </dependency>
  5. <dependency>
  6.     <groupId>org.springframework.boot</groupId>
  7.     <artifactId>spring‐boot‐starter‐test</artifactId>
  8. </dependency>

2、在application.yml中添加RabbitMQ的配置：  

  1. server:
  2.   port: 10086  
  3. spring:
  4.   application:
  5.     name: mq-rabbitmq-producer
  6.   rabbitmq:
  7.     host: 192.168.1.103
  8.     port: 5672
  9.     username: kavito
  10.     password: 123456
  11.     virtualHost: /kavito
  12.     template:
  13.       retry:
  14.         enabled: true
  15.         initial-interval: 10000ms
  16.         max-interval: 300000ms
  17.         multiplier: 2
  18.       exchange: topic.exchange
  19.     publisher-confirms: true

属性说明： 

  • template：有关AmqpTemplate的配置
    • retry：失败重试
      • enabled：开启失败重试
      • initial-interval：第一次重试的间隔时长
      • max-interval：最长重试间隔，超过这个间隔将不再重试
      • multiplier：下次重试间隔的倍数，此处是2即下次重试间隔是上次的2倍
    • exchange：缺省的交换机名称，此处配置后，发送消息如果不指定交换机就会使用这个
  • publisher-confirms：生产者确认机制，确保消息会正确发送，如果发送失败会有错误回执，从而触发重试

当然如果consumer只是接收消息而不发送，就不用配置template相关内容。  

 3、定义RabbitConfig配置类，配置Exchange、Queue、及绑定交换机。

  1. @Configuration
  2. public class RabbitmqConfig {
  3.     public static final String QUEUE_EMAIL = "queue_email";//email队列
  4.     public static final String QUEUE_SMS = "queue_sms";//sms队列
  5.     public static final String EXCHANGE_NAME="topic.exchange";//topics类型交换机
  6.     public static final String ROUTINGKEY_EMAIL="topic.#.email.#";
  7.     public static final String ROUTINGKEY_SMS="topic.#.sms.#";
  8.  
  9.     //声明交换机
  10.     @Bean(EXCHANGE_NAME)
  11.     public Exchange exchange(){
  12.         //durable(true) 持久化，mq重启之后交换机还在
  13.         return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
  14.     }
  15.  
  16.     //声明email队列
  17.     /*
  18.      *   new Queue(QUEUE_EMAIL,true,false,false)
  19.      *   durable="true" 持久化 rabbitmq重启的时候不需要创建新的队列
  20.      *   auto-delete 表示消息队列没有在使用时将被自动删除 默认是false
  21.      *   exclusive  表示该消息队列是否只在当前connection生效,默认是false
  22.      */
  23.     @Bean(QUEUE_EMAIL)
  24.     public Queue emailQueue(){
  25.         return new Queue(QUEUE_EMAIL);
  26.     }
  27.     //声明sms队列
  28.     @Bean(QUEUE_SMS)
  29.     public Queue smsQueue(){
  30.         return new Queue(QUEUE_SMS);
  31.     }
  32.  
  33.     //ROUTINGKEY_EMAIL队列绑定交换机，指定routingKey
  34.     @Bean
  35.     public Binding bindingEmail(@Qualifier(QUEUE_EMAIL) Queue queue,
  36.                                 @Qualifier(EXCHANGE_NAME) Exchange exchange){
  37.         return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_EMAIL).noargs();
  38.     }
  39.     //ROUTINGKEY_SMS队列绑定交换机，指定routingKey
  40.     @Bean
  41.     public Binding bindingSMS(@Qualifier(QUEUE_SMS) Queue queue,
  42.                               @Qualifier(EXCHANGE_NAME) Exchange exchange){
  43.         return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_SMS).noargs();
  44.     }
  45.  
  46. }

生产者(mq-rabbitmq-producer)

为了方便测试，我直接把生产者代码放工程测试类：发送routing key是"topic.sms.email"的消息，那么mq-rabbitmq-consumer下那些监听的（与交换机(topic.exchange)绑定，并且订阅的routingkey中匹配了"topic.sms.email"规则的） 队列就会收到消息。

  1. @SpringBootTest
  2. @RunWith(SpringRunner.class)
  3. public class Send {
  4.  
  5.     @Autowired
  6.     RabbitTemplate rabbitTemplate;
  7.     
  8.     @Test
  9.     public void sendMsgByTopics(){
  10.  
  11.         /**
  12.          * 参数：
  13.          * 1、交换机名称
  14.          * 2、routingKey
  15.          * 3、消息内容
  16.          */
  17.         for (int i=0;i<5;i++){
  18.             String message = "恭喜您，注册成功！userid="+i;
  19.             rabbitTemplate.convertAndSend(RabbitmqConfig.EXCHANGE_NAME,"topic.sms.email",message);
  20.             System.out.println(" [x] Sent '" + message + "'");
  21.         }
  22.  
  23.     }
  24. }

运行测试类发送5条消息：

 

web管理界面： 可以看到已经创建了交换机以及queue_email、queue_sms 2个队列，并且向这两个队列分别发送了5条消息

消费者(mq-rabbitmq-consumer)

编写一个监听器组件，通过注解配置消费者队列，以及队列与交换机之间绑定关系。（也可以像生产者那样通过配置类配置）

在SpringAmqp中，对消息的消费者进行了封装和抽象。一个JavaBean的方法，只要添加@RabbitListener注解，就可以成为了一个消费者。

  1. @Component
  2. public class ReceiveHandler {
  3.  
  4.     //监听邮件队列
  5.     @RabbitListener(bindings = @QueueBinding(
  6.             value = @Queue(value = "queue_email", durable = "true"),
  7.             exchange = @Exchange(
  8.                     value = "topic.exchange",
  9.                     ignoreDeclarationExceptions = "true",
  10.                     type = ExchangeTypes.TOPIC
  11.             ),
  12.             key = {"topic.#.email.#","email.*"}))
  13.     public void rece_email(String msg){
  14.         System.out.println(" [邮件服务] received : " + msg + "!");
  15.     }
  16.  
  17.     //监听短信队列
  18.     @RabbitListener(bindings = @QueueBinding(
  19.             value = @Queue(value = "queue_sms", durable = "true"),
  20.             exchange = @Exchange(
  21.                     value = "topic.exchange",
  22.                     ignoreDeclarationExceptions = "true",
  23.                     type = ExchangeTypes.TOPIC
  24.             ),
  25.             key = {"topic.#.sms.#"}))
  26.     public void rece_sms(String msg){
  27.         System.out.println(" [短信服务] received : " + msg + "!");
  28.     }
  29. }

属性说明： 

  • @Componet：类上的注解，注册到Spring容器
  • @RabbitListener：方法上的注解，声明这个方法是一个消费者方法，需要指定下面的属性：
    • bindings：指定绑定关系，可以有多个。值是@QueueBinding的数组。@QueueBinding包含下面属性：
      • value：这个消费者关联的队列。值是@Queue，代表一个队列
      • exchange：队列所绑定的交换机，值是@Exchange类型
      • key：队列和交换机绑定的RoutingKey，可指定多个

启动mq-rabbitmq-comsumer项目

ok,邮件服务和短息服务接收到消息后，就可以各自开展自己的业务了。 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小小林熬夜学编程/article/detail/594706
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号