RabbitMQ五种消息队列学习（三）--Work模式

你的名字 2022-06-03 02:51 481阅读 0赞

# RabbitMQ五种消息队列学习（三）–Work模式 #

标签（空格分隔）： RabbitMQ

--------------------

由于在实际应用中，简单队列模型无法解决很多实际问题，而且生产者和消费者是一对一的关系。模型较为单一。故引入Work模式。

## 结构图 ##

![这里写图片描述][SouthEast]

> 一个生产者、多个消费者。  
> **一个消息只能被一个消费者获取。**

测试实现：  
1、生产者

生产者循环创建消息  
2、消费者一

private final static String QUEUE_NAME = "test_queue_work";
    
        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.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);
            }
        }

3、消费者二  
注意：channel.basicQos(1); 被注释掉了

private final static String QUEUE_NAME = "test_queue_work";
    
        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.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 + "'");
                // 休眠1秒
                Thread.sleep(1000);
    
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }

### 测试结果： ###

1、 消费者1和消费者2获取到的消息内容是不同的，同一个消息只能被一个消费者获取。  
2、 消费者1和消费者2获取到的消息的数量是相同的，一个是奇数一个是偶数。

其实，这样是不合理的，应该是消费者1要比消费者2获取到的消息多才对。

## 消息的确认模式 ##

消费者从队列中获取消息，服务端如何知道消息已经被消费呢？

模式1：自动确认

> 只要消息从队列中获取，无论消费者获取到消息后是否成功消息，都认为是消息已经成功消费。

![这里写图片描述][SouthEast 1]

模式2：手动确认

> 消费者从队列中获取消息后，服务器会将该消息标记为不可用状态，等待消费者的反馈，如果消费者一直没有反馈，那么该消息将一直处于不可用状态。

![这里写图片描述][SouthEast 2]

### 应用场景 ###

> 集群中同时监听一个服务队列

[SouthEast]: /images/20220603/a62fa16483564a43b06322335337ae43.png
[SouthEast 1]: /images/20220603/89eccf1723e141988976aced069df722.png
[SouthEast 2]: /images/20220603/84a87ed78da947919f7b05a90437166e.png