通配符模式

一個(gè)消息發(fā)送者，發(fā)送消息
每個(gè)消息接收者，都已一個(gè)獨(dú)立的隊(duì)列

消息會發(fā)送到交換機(jī)，由交換機(jī)發(fā)送到隊(duì)列
根據(jù)key，模糊匹配，來接收消息

Send

發(fā)送者

package cn.itcast.rabbitmq.topic;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_topic";public static void main(String[] argv) throws Exception {// 獲取到連接以及mq通道Connection connection = ConnectionUtil.getConnection();Channel channel = connection.createChannel();// 聲明exchangechannel.exchangeDeclare(EXCHANGE_NAME, "topic");// 消息內(nèi)容String message = "刪除商品 ，id=1000";channel.basicPublish(EXCHANGE_NAME, "item.delete", null, message.getBytes());System.out.println(" [x] Sent '" + message + "'");channel.close();connection.close();} }

Recv

消費(fèi)者1

package cn.itcast.rabbitmq.topic;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_topic_topic1";private final static String EXCHANGE_NAME = "test_exchange_topic";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, "item.update");channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");// 同一時(shí)刻服務(wù)器只會發(fā)一條消息給消費(fèi)者channel.basicQos(1);// 定義隊(duì)列的消費(fèi)者QueueingConsumer consumer = new QueueingConsumer(channel);// 監(jiān)聽隊(duì)列，手動返回完成channel.basicConsume(QUEUE_NAME, false, consumer);// 獲取消息while (true) {QueueingConsumer.Delivery delivery = consumer.nextDelivery();String message = new String(delivery.getBody());System.out.println(" 前臺系統(tǒng)： '" + message + "'");Thread.sleep(10);channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}} }

Recv2

消費(fèi)者2

package cn.itcast.rabbitmq.topic;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_topic_topic2";private final static String EXCHANGE_NAME = "test_exchange_topic";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, "item.#");// 同一時(shí)刻服務(wù)器只會發(fā)一條消息給消費(fèi)者channel.basicQos(1);// 定義隊(duì)列的消費(fèi)者QueueingConsumer consumer = new QueueingConsumer(channel);// 監(jiān)聽隊(duì)列，手動返回完成channel.basicConsume(QUEUE_NAME, false, consumer);// 獲取消息while (true) {QueueingConsumer.Delivery delivery = consumer.nextDelivery();String message = new String(delivery.getBody());System.out.println(" 搜索系統(tǒng)： '" + message + "'");Thread.sleep(10);channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}} }

啟動發(fā)送者

啟動消費(fèi)者

測試

生產(chǎn)者，依次發(fā)送消息
Item.insert、item.update、item.delete

消費(fèi)者1
前臺系統(tǒng)，接收到2條消息

消費(fèi)者2
搜索系統(tǒng)，接收到3條消息

注意
通配符，指的是消費(fèi)者
而不是發(fā)送者，發(fā)送者不可以使用通配符

《新程序員》：云原生和全面數(shù)字化實(shí)踐50位技術(shù)專家共同創(chuàng)作，文字、視頻、音頻交互閱讀

總結(jié)

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

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