RabbitMQ3work模式

Posted 2021-01-08 eastday

    上一篇博客的作为rabbitMQ的入门程序，也是简单队列模式，一个生产者，一个消费者，今天这篇博客介绍work模式，一个生产者，多个消费者，下面的例子模拟两个消费者的情况。

 

图示：
        

 

  一个生产者、两个消费者；一个消息只能被一个消费者获取。

  在work模式中可以分为两种模式，一种是两个消费者平均消费队列中的消息，即使他们的消费能力是不一样的，这种似乎不太符合实际的情况。另一种是能者多劳模式，处理消息能力强的消费者会获取更多的 消息，这种模式更符合实际需求。

 

  生产者：向队列发送50条消息，下面代码中，生产者每生产一条消息后都会休眠一段时间，并且越往后休眠的时间越长。

 

public class Send {
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);
for (int i = 0; i < 50; i++) {
// 消息内容
String message = "" + i;
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent ‘" + message + "‘");
Thread.sleep(i * 10);
}
channel.close();
connection.close();
}
}

 

消费者1：休眠时间为10ms

 

public class Recv {
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);
}
}
}

 

消费者2：休眠时间为1000ms

 

 

 

public class Recv2 {
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);
}
}
}

 

  从上面代码中我们知道消费者2的休眠时间更长，消费者1的处理能力应该比消费者2更强，但是最后运行的结果会发现，两个消费者获得消息的数量十一样的，一个全部为奇数，一个全部为偶数。

  我们把每个消费者中的代码：channel.basicQos(1);这句代码解开注释会发现消费者2获取到8条数据，而消费者1获取到42条消息。这样 的结果才符合work模式的能者多劳模式。这句代码表示服务器同一时刻只会发送一条消息给消费者，但并不指定是哪个消费者，处理能力高的人会接收到更多的 消息。

 

小结：

  简单队列和work 模式的不同：
  简单队列只要消息从队列中获取，无论消费者获取到消息后是否成功消费，比如遇到状况：断电，都认为是消息已经成功消费；

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

  具体使用哪种模式具体问题具体分析。

  channel.basicConsume(QUEUE_NAME, false, consumer);false表示监听队列，手动返回完成状态，true表示自动返回。