目錄

?

?

基本概念

代碼與實例

---

?

基本概念

過程圖如下：

主要是把交換機設置為直連的方式direct直連的方式然后把 數據 發送給交換機。

交換機再通過路由的Key值轉發到隊列上。

每一個客戶端，都有一個隊列，監聽自己的隊列就可以收到RabbitMQ轉發過來的數據了。

?

代碼與實例

其中recv1就綁定了error路由鍵

recv2綁定了info、error、warning路由鍵

對應的RabbitMQ如下：

程序運行截圖如下：

當只發送error作為路由鍵時：

此時發送端如下：

接收端如下：

當發送端key值為info時！

接收端如下：

?

代碼如下：

Recv1.java

package routing;import com.rabbitmq.client.*; import util.ConnectionUtils;import java.io.IOException; import java.util.concurrent.TimeoutException;public class Recv1 {private static final String EXCHANGE_NAME = "test_exchange_direct";private static final String QUEUE_NAME = "test_queue_direct_1";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = ConnectionUtils.getConnect();final Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);channel.basicQos(1);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "error");Consumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String msg = new String(body, "utf-8");System.out.println("[1] Recv msg : " + msg);try {Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();} finally {System.out.println("[1] done");channel.basicAck(envelope.getDeliveryTag(), false);}}};boolean autoAck = false;channel.basicConsume(QUEUE_NAME, autoAck, consumer);} }

Recv2.java

package routing;import com.rabbitmq.client.*; import util.ConnectionUtils;import java.io.IOException; import java.util.concurrent.TimeoutException;public class Recv2 {private static final String EXCHANGE_NAME = "test_exchange_direct" +"";private static final String QUEUE_NAME = "test_queue_direct_2";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = ConnectionUtils.getConnect();final Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);channel.basicQos(1);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "error");channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info");channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "warning");Consumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String msg = new String(body, "utf-8");System.out.println("[2] Recv msg : " + msg);try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();} finally {System.out.println("[2] done");channel.basicAck(envelope.getDeliveryTag(), false);}}};boolean autoAck = false;channel.basicConsume(QUEUE_NAME, autoAck, consumer);} }

Send.java

package routing;import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import util.ConnectionUtils;import java.io.IOException; import java.util.concurrent.TimeoutException;public class Send {private static final String EXCHANGE_NAME = "test_exchange_direct";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = ConnectionUtils.getConnect();Channel channel = connection.createChannel();//聲明交換機channel.exchangeDeclare(EXCHANGE_NAME, "direct");String msg = "Hello direct!";String routingKey = "info";System.out.println("Send msg : " + msg);channel.basicPublish(EXCHANGE_NAME, routingKey, null, msg.getBytes());channel.close();connection.close();} }

?

源碼打包下載地址：

https://github.com/fengfanchen/Java/tree/master/Routing

創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎

總結

以上是生活随笔為你收集整理的Java笔记-使用RabbitMQ的Java接口实现Routing(路由模式)的全部內容，希望文章能夠幫你解決所遇到的問題。

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