上一篇博客的作为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表示自动返回。
原文地址:https://www.cnblogs.com/eastday/p/9692446.html