RabbitMQ的Work能者多劳模式

Work能者多劳模式
一个消息生成者
一个队列，多个消息消费者

一个消息，只能被一个消费接收

关键代码

// 同一时刻服务器只会发一条消息给消费者 channel.basicQos(1);

Send

生产者

package cn.itcast.rabbitmq.work;import cn.itcast.rabbitmq.util.ConnectionUtil;import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection;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();} }

Recv

消费者1

package cn.itcast.rabbitmq.work;import cn.itcast.rabbitmq.util.ConnectionUtil;import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.QueueingConsumer;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);}} }

Recv2

消费者2

package cn.itcast.rabbitmq.work;import cn.itcast.rabbitmq.util.ConnectionUtil;import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.QueueingConsumer;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);}} }

消费者1
接收消息，休眠10毫秒
消费者2
接收消息，休眠1000毫秒

测试结果

一共发送50条消息

消费者1
获取的多

消费者2
获取的少

总结

以上是生活随笔为你收集整理的RabbitMQ的Work能者多劳模式的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。