訂閱者模式

一個(gè)生產(chǎn)者，多個(gè)消費(fèi)者
每一個(gè)消費(fèi)者，都有一個(gè)獨(dú)立的隊(duì)列
生產(chǎn)者沒有將消息直接發(fā)送到隊(duì)列，而是發(fā)送到了交換機(jī)

每個(gè)隊(duì)列都要綁定到交換機(jī)
生產(chǎn)者發(fā)送的消息，經(jīng)過交換機(jī)，到達(dá)隊(duì)列
實(shí)現(xiàn)，一個(gè)消息被多個(gè)消費(fèi)者獲取的目的

注意
消息發(fā)送到?jīng)]有隊(duì)列綁定的交換機(jī)時(shí)，消息將丟失
因?yàn)?#xff0c;交換機(jī)沒有存儲(chǔ)消息的能力，消息只能存在在隊(duì)列中

Send

生產(chǎn)者

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");// 消息內(nèi)容String message = "消息已經(jīng)修改，商品id=1000";channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());System.out.println(" [x] Sent '" + message + "'");channel.close();connection.close();} }

Recv

消費(fèi)者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();// 聲明隊(duì)列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 綁定隊(duì)列到交換機(jī)channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");// 同一時(shí)刻服務(wù)器只會(huì)發(fā)一條消息給消費(fèi)者channel.basicQos(1);// 定義隊(duì)列的消費(fèi)者QueueingConsumer consumer = new QueueingConsumer(channel);// 監(jiān)聽隊(duì)列，手動(dòng)返回完成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

消費(fèi)者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();// 聲明隊(duì)列channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 綁定隊(duì)列到交換機(jī)channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");// 同一時(shí)刻服務(wù)器只會(huì)發(fā)一條消息給消費(fèi)者channel.basicQos(1);// 定義隊(duì)列的消費(fèi)者QueueingConsumer consumer = new QueueingConsumer(channel);// 監(jiān)聽隊(duì)列，手動(dòng)返回完成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);}} }

啟動(dòng)發(fā)送者

此時(shí)
沒有綁定隊(duì)列

把消息發(fā)送到，沒有綁定隊(duì)列的交換機(jī)時(shí)
消息將會(huì)丟失

啟動(dòng)兩個(gè)消費(fèi)者

測(cè)試

生產(chǎn)者，發(fā)送消息
消費(fèi)者1，拿到了消息

消費(fèi)者2，拿到了消息

同一個(gè)消息
可以被多個(gè)消費(fèi)者獲取

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。