路由模式

一個生產(chǎn)者，發(fā)送消息
每個消費者，都有一個獨立的隊列

消息發(fā)送到交換機，交換機發(fā)送到每個隊列
根據(jù)key，是否相等，來接收消息

Send

生產(chǎn)者

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

Recv

消費者1

package cn.itcast.rabbitmq.routing;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_direct1";private final static String EXCHANGE_NAME = "test_exchange_direct";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, "update");channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");// 同一時刻服務器只會發(fā)一條消息給消費者channel.basicQos(1);// 定義隊列的消費者QueueingConsumer consumer = new QueueingConsumer(channel);// 監(jiān)聽隊列，手動返回完成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

消費者2

package cn.itcast.rabbitmq.routing;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_direct2";private final static String EXCHANGE_NAME = "test_exchange_direct";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, "insert");channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");// 同一時刻服務器只會發(fā)一條消息給消費者channel.basicQos(1);// 定義隊列的消費者QueueingConsumer consumer = new QueueingConsumer(channel);// 監(jiān)聽隊列，手動返回完成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ā)送者

啟動消費者

此時，有五個綁定隊列

消費者1，兩個
消費者2，三個

測試

發(fā)送消息，依次將key設置為
insert、update、delete

消費者1
前臺系統(tǒng)，接收了2條消息

消費者2
搜索系統(tǒng)，接收了3條消息

總結(jié)

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

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