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

这里写图片描述

  1. 一个生产者、多个消费者。 
  2. 一个消息只能被一个消费者获取。

测试实现： 
1、生产者

  1.  private final static String QUEUE_NAME = "test_queue_work";
  3.     public static void main(String[] argv) throws Exception {
  4.         // 获取到连接以及mq通道
  5.         Connection connection = ConnectionUtil.getConnection();
  6.         Channel channel = connection.createChannel();
  8.         // 声明队列
  9.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
  11.         for (int i = 0; i < 50; i++) {
  12.             // 消息内容
  13.             String message = "" + i;
  14.             channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
  15.             System.out.println(" [x] Sent \'" + message + "\'");
  17.             Thread.sleep(i * 10);
  18.         }
  20.         channel.close();
  21.         connection.close();
  22.     }

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

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

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

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

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

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

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

模式1：自动确认

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

模式2：手动确认

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

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

 

版权声明：本文为shanheyongmu原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://www.cnblogs.com/shanheyongmu/p/10406155.html