RabbitMQ五种消息队列学习（六）--通配符模式（路由类型：Topic）

淩亂°似流年 2022-06-03 03:59 251阅读 0赞

RabbitMQ五种消息队列学习（六）–通配符模式（路由类型：Topic）

标签（空格分隔）： RabbitMQ

正如上一篇文章中所描述的一种模式的升级，如果需要监听某个交换机的所有消息的话，可以通过消息队列的形式进行绑定。

队列结构图

这里写图片描述
通过字符串通配符的模式匹配需要分发的消息队列
消息路径：
这里写图片描述

实现：
1、生产者

  1. private final static String EXCHANGE_NAME = "test_exchange_topic";
  3.     public static void main(String[] argv) throws Exception {
  4.         // 获取到连接以及mq通道
  5.         Connection connection = ConnectionUtil.getConnection();
  6.         Channel channel = connection.createChannel();
  8.         // 声明exchange
  9.         channel.exchangeDeclare(EXCHANGE_NAME, "topic");
  11.         // 消息内容
  12.         String message = "删除商品 id = 1000";
  13.         channel.basicPublish(EXCHANGE_NAME, "item.delete", null, message.getBytes());
  14.         System.out.println(" [x] Sent '" + message + "'");
  16.         channel.close();
  17.         connection.close();
  18.     }

2、消费者1

  1. private final static String QUEUE_NAME = "test_queue_topic_1";
  3.     private final static String EXCHANGE_NAME = "test_exchange_topic";
  5.     public static void main(String[] argv) throws Exception {
  7.         // 获取到连接以及mq通道
  8.         Connection connection = ConnectionUtil.getConnection();
  9.         Channel channel = connection.createChannel();
  11.         // 声明队列
  12.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  14.         // 绑定队列到交换机
  15.         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update");
  16.         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");
  18.         // 同一时刻服务器只会发一条消息给消费者
  19.         channel.basicQos(1);
  21.         // 定义队列的消费者
  22.         QueueingConsumer consumer = new QueueingConsumer(channel);
  23.         // 监听队列，手动返回完成
  24.         channel.basicConsume(QUEUE_NAME, false, consumer);
  26.         // 获取消息
  27.         while (true) {
  28.             QueueingConsumer.Delivery delivery = consumer.nextDelivery();
  29.             String message = new String(delivery.getBody());
  30.             System.out.println(" [x] Received '" + message + "'");
  31.             Thread.sleep(10);
  33.             channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
  34.         }
  35.     }

3、消费者2

  1. private final static String QUEUE_NAME = "test_queue_topic_2";
  3.     private final static String EXCHANGE_NAME = "test_exchange_topic";
  5.     public static void main(String[] argv) throws Exception {
  7.         // 获取到连接以及mq通道
  8.         Connection connection = ConnectionUtil.getConnection();
  9.         Channel channel = connection.createChannel();
  11.         // 声明队列
  12.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  14.         // 绑定队列到交换机
  15.         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.#");
  17.         // 同一时刻服务器只会发一条消息给消费者
  18.         channel.basicQos(1);
  20.         // 定义队列的消费者
  21.         QueueingConsumer consumer = new QueueingConsumer(channel);
  22.         // 监听队列，手动返回完成
  23.         channel.basicConsume(QUEUE_NAME, false, consumer);
  25.         // 获取消息
  26.         while (true) {
  27.             QueueingConsumer.Delivery delivery = consumer.nextDelivery();
  28.             String message = new String(delivery.getBody());
  29.             System.out.println(" [x] Received '" + message + "'");
  30.             Thread.sleep(10);
  32.             channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
  33.         }
  34.     }

4、测试结果
消费者1中，明确了队列需要绑定路由器的哪些Routing Key

  1. // 绑定队列到交换机
  2.         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update");
  3.         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");

消费者2，因为要接收item下的所有消息，所以绑定为：

  1. // 绑定队列到交换机
  2.         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.#");

所有消费者2可以接收所有消息，消费者1只能接受 update\delete两种消息

发表评论

表情：
评论列表 （有 0 条评论，251人围观）

还没有评论，来说两句吧...

相关阅读