• RabbitMQ初步到精通-第四章-RabbitMQ工作模式-Routing

    第四章-RabbitMQ工作模式-Routing

    1.模式介绍

    1.1 模式

    路由模式-继续还是和Exchange打交道，上节提到的Exchange类型为Fanout，此次声明的类型为direct 与默认Exchange一致。但还有最核心的一点，上节未使用Routing key，此次模式中，会依赖于此Key。

    如下图：

    消费发送的时候，会指定routing key ，例如指定 black，那么消费发送至 X -则会根据routing key路由消息至队列Q2，被消费者C2进行消费。

    若此时有个key ，比如blue 同时指向了 Q1 和 Q2，那消息也会同时传送到Q1与Q2，这与广播模式很像了。

    1.2 场景

    业务中需要根据不同的条件对所产生的消息进行不同的独立消费，可以使用此模式。

    例如系统中产生不同level的日志，根据level的不同上送到不同的queue，进行相应的消费。

    info类型指定 info的queue，debug类型指定debug的queue，error类型的指定error的queue

    实际运用不多，有实际生产运用的小伙伴可以留言，学习下。

    1.3 模拟

    三个重点 1.声明direct类型的 交换机 2. 绑定好交换机与queue 3. 发消息指定路由键

    2.代码验证

    梦总会醒的，小明觉得复制水，不现实，搞个现实的开关吧。因为有天小丽对他说，洗水澡很没意思，想洗牛奶澡，要小明一定帮她实现。小明自己想了想觉得可以试试，但自己也想搞个新奇的，不喜欢牛奶，喜欢红酒，洗个红酒澡吧，一边洗，一边喝。醉洗 -- o(*￣︶￣*)o。

    2.1 生产者

    1.  
    2. /**
    3.  * @author rabbit
    4.  * @version 1.0.0
    5.  * @Description 一个生产者，一个交换机，两个队列，两个消费者
    6.  * 声明一个DIRECT类型的exchange，并且根据Routingkey绑定指定的队列
    7.  * 生产者在创建DIRECT类型的exchange后绑定相应的队列，并且指定Routingkey。在发送消息是也要指定消息的Routingkey
    8.  * 
    9.  * 另：TOPIC模式
    10.  * 生产者创建Topic的exchange并且并且指定队列，这次绑定可以通过*和#匹配关键字，对指定RoutingKey内容进行匹配。
    11.  * *（星号）可以代替一个单词。
    12.  * ＃（哈希）可以替代零个或多个单词。
    13.  * channel.queueBind("topics-queue-1", "topics-exchange", "zhang.*");
    14.  * channel.basicPublish("topics-exchange", "zhang.sna", null, "张三".getBytes());
    15.  * @createTime 2022/07/27 19:34:00
    16.  */
    17. public class WaterProducer {
    18.     public static final String PUBSUB_QUEUE_1 = "SolarWaterHeater-RedWine";
    19.  
    20.     public static final String PUBSUB_QUEUE_2 = "SolarWaterHeater-Milk";
    21.  
    22.  
    23.  
    24.     //生产者
    25.     public static void main(String[] args) throws Exception {
    26.         //1、获取connection
    27.         Connection connection = RabbitCommonConfig.getConnection();
    28.         //2、创建channel
    29.         Channel channel = connection.createChannel();
    30.  
    31.         for (int i = 1; i <= 10; i++) {
    32.             sendMsg(channel, i);
    33.             Thread.sleep(1000);
    34.         }
    35.  
    36.         //4、关闭管道和连接
    37.         channel.close();
    38.         connection.close();
    39.     }
    40.  
    41.     private static void sendMsg(Channel channel, int k) throws IOException {
    42.  
    43.         //3、创建exchange并且指定类型
    44.         channel.exchangeDeclare("routing-exchange", BuiltinExchangeType.DIRECT);
    45.  
    46.         //4、绑定队列 routing-queue-error routing-queue-info
    47.         channel.queueBind(PUBSUB_QUEUE_2, "routing-exchange", "MILK");
    48.         channel.queueBind(PUBSUB_QUEUE_1, "routing-exchange", "REDWINE");
    49.  
    50.         /**
    51.          * 参数1：指定exchange，使用“”。默认的exchange
    52.          * 参数2：指定路由的规则，使用具体的队列名称。exchange为""时，消息直接发送到队列中
    53.          * 参数3：指定传递的消息携带的properties
    54.          * 参数4：指定传递的消息，byte[]类型
    55.          */
    56.  
    57.         //5、发送消息并且指定接收的队列的routingkey
    58.         channel.basicPublish("routing-exchange", "MILK", null, ("牛奶-" + k+"升").getBytes());
    59.         channel.basicPublish("routing-exchange", "REDWINE", null, ("红酒-" + k+"升").getBytes());
    60.  
    61.         System.out.println("水龙头分别放牛奶和红酒成功！" + k + "升");
    62.     }
    63.  
    64. }

    2.2 消费者

    小明：

    1.  
    2. /**
    3.  * @author rabbit
    4.  * @version 1.0.0
    5.  * @Description Routing 模式 一个生产者，一个交换机，两个队列，两个消费者
    6.  * @createTime 2022/07/27 19:36:00
    7.  */
    8. public class XMShowerConsumer {
    9.     public static final String PUBSUB_QUEUE_1 = "SolarWaterHeater-RedWine";
    10.  
    11.     //消费者
    12.     public static void consumer() throws Exception {
    13.         //1、获取连对象、
    14.         Connection connection = RabbitCommonConfig.getConnection();
    15.         //2、创建channel
    16.         Channel channel = connection.createChannel();
    17.  
    18.         channel.basicQos(1);
    19.         //3、创建队列-helloworld
    20.         /**
    21.          * 参数1：queue 指定队列名称
    22.          * 参数2：durable 是否开启持久化（true）
    23.          * 参数3：exclusive 是否排外（conn.close()-》当前对列自动删除，当前队列只能被一个 消费者消费）
    24.          * 参数4：autoDelete 如果这个队列没有其他消费者在消费，队列自动删除
    25.          * 参数5：arguments 指定队列携带的信息
    26.          */
    27.         channel.queueDeclare(PUBSUB_QUEUE_1, true, false, false, null);
    28.  
    29.         //4.开启监听Queue
    30.         DefaultConsumer consumer = new DefaultConsumer(channel) {
    31.             @Override
    32.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    33.                 System.out.println("小明洗澡用: " + new String(body, "UTF-8"));
    34.                 try {
    35.                     Thread.sleep(1000);
    36.                 } catch (InterruptedException e) {
    37.                     e.printStackTrace();
    38.                 }
    39.                 //手动ACK（接收信息，指定是否批量操作）
    40.                 channel.basicAck(envelope.getDeliveryTag(), false);
    41.             }
    42.         };
    43.         /**
    44.          * 参数1：queue 指定消费哪个队列
    45.          * 参数1：deliverCallback 指定是否ACK（true:收到消息会立即告诉RabbiMQ，false：手动告诉）
    46.          * 参数1：cancelCallback 指定消费回调
    47.          *
    48.          */
    49.         //3.关闭自动ACK
    50.         channel.basicConsume(PUBSUB_QUEUE_1, false, consumer);
    51.         System.out.println("小明开始洗红酒澡......");
    52.  
    53.         //5、键盘录入，让程序不结束！
    54.         System.in.read();
    55.  
    56.         //6、释放资源
    57.         channel.close();
    58.         connection.close();
    59.     }
    60.  
    61.     public static void main(String[] args) throws Exception {
    62.         XMShowerConsumer.consumer();
    63.     }
    64.  
    65. }

    小丽：

    1.  
    2. /**
    3.  * @author rabbit
    4.  * @version 1.0.0
    5.  * @Description Routing 模式 一个生产者，一个交换机，两个队列，两个消费者
    6.  * @createTime 2022/07/27 19:36:00
    7.  */
    8. public class XLShowerConsumer {
    9.  
    10.     public static final String PUBSUB_QUEUE_2 = "SolarWaterHeater-Milk";
    11.  
    12.     //消费者
    13.     public static void consumer() throws Exception {
    14.         //1、获取连对象、
    15.         Connection connection = RabbitCommonConfig.getConnection();
    16.         //2、创建channel
    17.         Channel channel = connection.createChannel();
    18.  
    19.         channel.basicQos(1);
    20.         //3、创建队列-helloworld
    21.         /**
    22.          * 参数1：queue 指定队列名称
    23.          * 参数2：durable 是否开启持久化（true）
    24.          * 参数3：exclusive 是否排外（conn.close()-》当前对列自动删除，当前队列只能被一个 消费者消费）
    25.          * 参数4：autoDelete 如果这个队列没有其他消费者在消费，队列自动删除
    26.          * 参数5：arguments 指定队列携带的信息
    27.          */
    28.         channel.queueDeclare(PUBSUB_QUEUE_2, true, false, false, null);
    29.  
    30.         //4.开启监听Queue
    31.         DefaultConsumer consumer = new DefaultConsumer(channel) {
    32.             @Override
    33.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    34.                 System.out.println("小丽洗澡用: " + new String(body, "UTF-8"));
    35.                 try {
    36.                     Thread.sleep(1000);
    37.                 } catch (InterruptedException e) {
    38.                     e.printStackTrace();
    39.                 }
    40.                 //手动ACK（接收信息，指定是否批量操作）
    41.                 channel.basicAck(envelope.getDeliveryTag(), false);
    42.             }
    43.         };
    44.         /**
    45.          * 参数1：queue 指定消费哪个队列
    46.          * 参数1：deliverCallback 指定是否ACK（true:收到消息会立即告诉RabbiMQ，false：手动告诉）
    47.          * 参数1：cancelCallback 指定消费回调
    48.          *
    49.          */
    50.         //3.关闭自动ACK
    51.         channel.basicConsume(PUBSUB_QUEUE_2, false, consumer);
    52.         System.out.println("小丽开始洗牛奶澡......");
    53.  
    54.         //5、键盘录入，让程序不结束！
    55.         System.in.read();
    56.  
    57.         //6、释放资源
    58.         channel.close();
    59.         connection.close();
    60.     }
    61.  
    62.     public static void main(String[] args) throws Exception {
    63.         XLShowerConsumer.consumer();
    64.     }
    65.  
    66. }

    2.3 结论验证

    生产者：

    1. 水龙头分别放牛奶和红酒成功！1升
    2. 水龙头分别放牛奶和红酒成功！2升
    3. 水龙头分别放牛奶和红酒成功！3升
    4. 水龙头分别放牛奶和红酒成功！4升
    5. 水龙头分别放牛奶和红酒成功！5升
    6. 水龙头分别放牛奶和红酒成功！6升
    7. 水龙头分别放牛奶和红酒成功！7升
    8. 水龙头分别放牛奶和红酒成功！8升
    9. 水龙头分别放牛奶和红酒成功！9升
    10. 水龙头分别放牛奶和红酒成功！10升

    小明：

    1. 小明洗澡用: 红酒-1升
    2. 小明洗澡用: 红酒-2升
    3. 小明洗澡用: 红酒-3升
    4. 小明洗澡用: 红酒-4升
    5. 小明洗澡用: 红酒-5升
    6. 小明洗澡用: 红酒-6升
    7. 小明洗澡用: 红酒-7升
    8. 小明洗澡用: 红酒-8升
    9. 小明洗澡用: 红酒-9升
    10. 小明洗澡用: 红酒-10升

    小丽

    1. 小丽洗澡用: 牛奶-1升
    2. 小丽洗澡用: 牛奶-2升
    3. 小丽洗澡用: 牛奶-3升
    4. 小丽洗澡用: 牛奶-4升
    5. 小丽洗澡用: 牛奶-5升
    6. 小丽洗澡用: 牛奶-6升
    7. 小丽洗澡用: 牛奶-7升
    8. 小丽洗澡用: 牛奶-8升
    9. 小丽洗澡用: 牛奶-9升
    10. 小丽洗澡用: 牛奶-10升

    通过开关控制，分别给热水器里注入红酒和牛奶，红酒和牛奶存储到各自独立的水槽中，供两个人洗澡使用。

    3.总结

    核心点：

    3.1 声明Direct 类型的Exchange

    3.2 绑定好对应的 Queue

    3.3 发送消息指定 Routing key

    核心代码：

    channel.exchangeDeclare("routing-exchange", BuiltinExchangeType.DIRECT);
    channel.queueBind(PUBSUB_QUEUE_2, "routing-exchange", "MILK");
    channel.basicPublish("routing-exchange", "MILK", null, ("牛奶-" + k+"升").getBytes());

    另：TOPIC模式

    topic模式不再单独展开

    topic模式最核心的内容是引入了通配符 * ，# 

    * *（星号）可以代替一个单词。 
* ＃（哈希）可以替代零个或多个单词。

    另外 声明的 Exchage 类型为topic 。

    看下模拟情况：

    1.首先声明Topic类型的 交换机

    2.使用带通配符的 routing key 做好绑定关系

    3. 发消息时，指定好对应的routing key ，按预期会将消息流转到对应的queue中。

  • 相关阅读:
    Python+Pytest+Allure+Yaml+Jenkins+GitLab接口自动化测试框架详解
    ensp实操浮动静态路由
    Pinpoint--基础--02--架构设计
    分布式数据库Mongodb——实验一
    UE5中一机一码功能
    Hello World分析
    OpenMP 入门
    web前端期末大作业——HTML+CSS+JavaScript仿王者荣耀游戏网站设计与制作
    图像超分综述:超长文一网打尽图像超分的前世今生 (附核心代码)
    CPU的睿频、超线程、SIMD指令集等特性对密码算法性能的影响
  • 原文地址：https://blog.csdn.net/blucastle/article/details/127929796