RabbitMQ---订阅模型-Direct

news2024/12/24 3:18:59

1、订阅模型-Direct

• 有选择性的接收消息
• 在订阅模式中，生产者发布消息，所有消费者都可以获取所有消息。
• 在路由模式中，我们将添加一个功能 - 我们将只能订阅一部分消息。
例如，我们只能将重要的错误消息引导到日志文件（以节省磁盘空间），同时仍然能够在控制台上打印所有日志消息。
• 但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
• 在Direct模型下，队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
• 消息的发送方在向Exchange发送消息时，也必须指定消息的routing key。
在这里插入图片描述

• P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
• X：Exchange（交换机），接收生产者的消息，然后把消息递交给与routing key完全匹配的队列
• C1：消费者，其所在队列指定了需要routing key 为 error 的消息
• C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

1.1、生产者

此处我们模拟商品的增删改，发送消息的RoutingKey分别是：insert、update、delete

public class Send {
   private final static String EXCHANGE_NAME = "direct_exchange_test";
   public static void main(String[] argv) throws Exception {
       // 获取到连接
       Connection connection = ConnectionUtil.getConnection();
       // 获取通道
       Channel channel = connection.createChannel();
       // 声明exchange，指定类型为direct
       channel.exchangeDeclare(EXCHANGE_NAME, "direct");
       // 消息内容
       String message = "商品新增了， id = 1001";
       // 发送消息，并且指定routing key 为：insert ,代表新增商品
       channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes());
       System.out.println(" [商品服务：] Sent '" + message + "'");
       channel.close();
       connection.close();
   }
}

1.2、消费者1

我们此处假设消费者1只接收两种类型的消息：更新商品和删除商品。

public class Recv {
   private final static String QUEUE_NAME = "direct_exchange_queue_1";
   private final static String EXCHANGE_NAME = "direct_exchange_test";
   public static void main(String[] argv) throws Exception {
       // 获取到连接
       Connection connection = ConnectionUtil.getConnection();
       // 获取通道
       Channel channel = connection.createChannel();
       // 声明队列
       channel.queueDeclare(QUEUE_NAME, false, false, false, null);
       // 绑定队列到交换机，同时指定需要订阅的routing key。假设此处需要update和delete消息
       channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
       channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
       // 定义队列的消费者
       DefaultConsumer consumer = new DefaultConsumer(channel) {
           // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
           @Override
           public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                   byte[] body) throws IOException {
               // body 即消息体
               String msg = new String(body);
               System.out.println(" [消费者1] received : " + msg + "!");
           }
       };
       // 监听队列，自动ACK
       channel.basicConsume(QUEUE_NAME, true, consumer);
   }
}

1.3、消费者2

我们此处假设消费者2接收所有类型的消息：新增商品，更新商品和删除商品。

public class Recv2 {
   private final static String QUEUE_NAME = "direct_exchange_queue_2";
   private final static String EXCHANGE_NAME = "direct_exchange_test";
   public static void main(String[] argv) throws Exception {
       // 获取到连接
       Connection connection = ConnectionUtil.getConnection();
       // 获取通道
       Channel channel = connection.createChannel();
       // 声明队列
       channel.queueDeclare(QUEUE_NAME, false, false, false, null);
       // 绑定队列到交换机，同时指定需要订阅的routing key。订阅 insert、update、delete
       channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");
       channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
       channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
       // 定义队列的消费者
       DefaultConsumer consumer = new DefaultConsumer(channel) {
           // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
           @Override
           public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                   byte[] body) throws IOException {
               // body 即消息体
               String msg = new String(body);
               System.out.println(" [消费者2] received : " + msg + "!");
           }
       };
       // 监听队列，自动ACK
       channel.basicConsume(QUEUE_NAME, true, consumer);
   }
}