Work模式

一個生產者、多個消費者
多個消費者，共同監聽一個隊列

一個消息，只能被一個消費者獲取

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毫秒

測試結果

消費者1
獲取了25條

消費者2
獲取了25條

消費者1，消費者2
獲得了相同數量的消息

總結

以上是生活随笔為你收集整理的RabbitMQ的Work模式的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。