訂閱者模式

一個生產者，多個消費者
每一個消費者，都有一個獨立的隊列
生產者沒有將消息直接發送到隊列，而是發送到了交換機

每個隊列都要綁定到交換機
生產者發送的消息，經過交換機，到達隊列
實現，一個消息被多個消費者獲取的目的

注意
消息發送到沒有隊列綁定的交換機時，消息將丟失
因為，交換機沒有存儲消息的能力，消息只能存在在隊列中

Send

生產者

package cn.itcast.rabbitmq.ps;import cn.itcast.rabbitmq.util.ConnectionUtil;import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection;public class Send {private final static String EXCHANGE_NAME = "test_exchange_fanout";public static void main(String[] argv) throws Exception {// 獲取到連接以及mq通道Connection connection = ConnectionUtil.getConnection();Channel channel = connection.createChannel();// 聲明exchangechannel.exchangeDeclare(EXCHANGE_NAME, "fanout");// 消息內容String message = "消息已經修改，商品id=1000";channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());System.out.println(" [x] Sent '" + message + "'");channel.close();connection.close();} }

Recv

消費者1

package cn.itcast.rabbitmq.ps;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";private final static String EXCHANGE_NAME = "test_exchange_fanout";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.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");// 同一時刻服務器只會發一條消息給消費者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.ps;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_work2";private final static String EXCHANGE_NAME = "test_exchange_fanout";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.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");// 同一時刻服務器只會發一條消息給消費者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);}} }

啟動發送者

此時
沒有綁定隊列

把消息發送到，沒有綁定隊列的交換機時
消息將會丟失

啟動兩個消費者

測試

生產者，發送消息
消費者1，拿到了消息

消費者2，拿到了消息

同一個消息
可以被多個消費者獲取

總結

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

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