RabbitMQ订阅者模式

ゝ一纸荒年。 2022-05-16 04:39 262阅读 0赞

### 订阅者模式 ###

一个生产者，多个消费者  
每一个消费者，都有一个独立的队列  
生产者没有将消息直接发送到队列，而是发送到了交换机

每个队列都要绑定到交换机  
生产者发送的消息，经过交换机，到达队列  
实现，一个消息被多个消费者获取的目的

注意  
消息发送到没有队列绑定的交换机时，消息将丢失  
因为，交换机没有存储消息的能力，消息只能存在在队列中  
![这里写图片描述][70]

### Send ###

生产者

package cn.itcast.rabbitmq.ps;
    
    import cn.itcast.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    public class Send {
    
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        public static void main(String[] argv) throws Exception {
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
    
            // 消息内容
            String message = "消息已经修改，商品id=1000";
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
    
            channel.close();
            connection.close();
        }
    }

### Recv ###

消费者1

package cn.itcast.rabbitmq.ps;
    
    import cn.itcast.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class Recv { 
    
        private final static String QUEUE_NAME = "test_queue_work";
    
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
            // 同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列，手动返回完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [x] Received '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

### Recv2 ###

消费者2

package cn.itcast.rabbitmq.ps;
    
    import cn.itcast.rabbitmq.util.ConnectionUtil;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class Recv2 { 
    
        private final static String QUEUE_NAME = "test_queue_work2";
    
        private final static String EXCHANGE_NAME = "test_exchange_fanout";
    
        public static void main(String[] argv) throws Exception {
    
            // 获取到连接以及mq通道
            Connection connection = ConnectionUtil.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
            // 同一时刻服务器只会发一条消息给消费者
            channel.basicQos(1);
    
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列，手动返回完成
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
            // 获取消息
            while (true) {
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                String message = new String(delivery.getBody());
                System.out.println(" [x] Received '" + message + "'");
                Thread.sleep(10);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }
    }

### 启动发送者 ###

![这里写图片描述][70 1]  
此时  
没有绑定队列  
![这里写图片描述][70 2]  
把消息发送到，没有绑定队列的交换机时  
消息将会丢失

### 启动两个消费者 ###

![这里写图片描述][70 3]

### 测试 ###

生产者，发送消息  
消费者1，拿到了消息  
![这里写图片描述][70 4]  
消费者2，拿到了消息  
![这里写图片描述][70 5]  
同一个消息  
可以被多个消费者获取

[70]: /images/20220516/2bbc742cda3b4553af225cf8d57f25dd.png
[70 1]: /images/20220516/c3cb460252454b96a132e57304b49bf6.png
[70 2]: /images/20220516/34e022971b52484199d0e4c3143ff067.png
[70 3]: /images/20220516/53b3b79fd0a241deb4bfd043b5e26db4.png
[70 4]: /images/20220516/f06acefc968e442082f63f724c16d61c.png
[70 5]: /images/20220516/eb1a1b8b3a3b46f6b3b68acacf2ebfff.png