RabbitMQ3work模式

Posted eastday

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RabbitMQ3work模式相关的知识,希望对你有一定的参考价值。

    上一篇博客的作为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表示自动返回。

 

以上是关于RabbitMQ3work模式的主要内容,如果未能解决你的问题,请参考以下文章

尝试使用片段保存夜间模式状态

是否有在单个活动中处理多个片段的 Android 设计模式?

csharp C#代码片段 - 使类成为Singleton模式。 (C#4.0+)https://heiswayi.github.io/2016/simple-singleton-pattern-us

常用代码片段

常用代码片段

使用 Git 来管理 Xcode 中的代码片段