2. 程式人生 > >RabbitMQ 的路由模式 Topic模式

RabbitMQ 的路由模式 Topic模式

阿新 • • 發佈：2018-12-05

模型

生產者

 1 package cn.wh;
 2 
 3 import java.io.IOException;
 4 import java.util.concurrent.TimeoutException;
 5 
 6 import cn.util.RabbitMqConnectionUtil;
 7 
 8 import com.rabbitmq.client.Channel;
 9 import com.rabbitmq.client.Connection;
10 
11 public class Send {
12     
13     private
 
 static final String EXCHANGE_NAME="test_exchange_direct";
14     
15     public static void main(String[] args) throws IOException, TimeoutException {
16         
17         
18         Connection connection = RabbitMqConnectionUtil.getConnection();
19         
20         Channel channel = connection.createChannel();

 
21         
22         //exchange
23         channel.exchangeDeclare(EXCHANGE_NAME, "direct");
24         
25         String  msg="hello direct!";
26         
27         
28         String routingKey="error";
29         channel.basicPublish(EXCHANGE_NAME, routingKey, null, msg.getBytes());
30         
31
 
         System.out.println("send "+msg);
32         
33         channel.close();
34         connection.close();
35     }
36 }

消費者

 

 1 package cn.wh;
 2 
 3 import java.io.IOException;
 4 import java.util.concurrent.TimeoutException;
 5 
 6 
 7 import cn.util.RabbitMqConnectionUtil;
 8 import com.rabbitmq.client.Channel;
 9 import com.rabbitmq.client.Connection;
10 import com.rabbitmq.client.Consumer;
11 import com.rabbitmq.client.DefaultConsumer;
12 import com.rabbitmq.client.Envelope;
13 import com.rabbitmq.client.AMQP.BasicProperties;
14 
15 public class Recv1 {
16     private static final String EXCHANGE_NAME = "test_exchange_direct";
17     private static final String QUEUE_NAME = "test_queue_direct_1";
18 
19     public static void main(String[] args) throws IOException, TimeoutException {
20 
21         Connection connection = RabbitMqConnectionUtil.getConnection();
22         final Channel channel = connection.createChannel();
23 
24         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
25         
26     
27         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "error");
28         
29         channel.basicQos(1);
30         
31         //定義一個消費者
32         Consumer consumer=new DefaultConsumer(channel){
33             //訊息到達 觸發這個方法
34             @Override
35             public void handleDelivery(String consumerTag, Envelope envelope,
36                     BasicProperties properties, byte[] body) throws IOException {
37              
38                 String msg=new String(body,"utf-8");
39                 System.out.println("[1] Recv msg:"+msg);
40                 
41                 try {
42                     Thread.sleep(2000);
43                 } catch (InterruptedException e) {
44                     e.printStackTrace();
45                 }finally{
46                     System.out.println("[1] done ");
47                     channel.basicAck(envelope.getDeliveryTag(), false);
48                 }
49             }
50         };
51         
52         boolean autoAck=false;//自動應答 false
53         channel.basicConsume(QUEUE_NAME,autoAck , consumer);
54     }
55         
56 }

消費者2

 1 package cn.wh;
 2 
 3 import java.io.IOException;
 4 import java.util.concurrent.TimeoutException;
 5 
 6 
 7 import cn.util.RabbitMqConnectionUtil;
 8 import com.rabbitmq.client.Channel;
 9 import com.rabbitmq.client.Connection;
10 import com.rabbitmq.client.Consumer;
11 import com.rabbitmq.client.DefaultConsumer;
12 import com.rabbitmq.client.Envelope;
13 import com.rabbitmq.client.AMQP.BasicProperties;
14 
15 public class Recv2 {
16     private static final String EXCHANGE_NAME = "test_exchange_direct";
17     private static final String QUEUE_NAME = "test_queue_direct_2";
18 
19     public static void main(String[] args) throws IOException, TimeoutException {
20 
21         Connection connection = RabbitMqConnectionUtil.getConnection();
22         final Channel channel = connection.createChannel();
23 
24         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
25         
26     
27         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "error");
28         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info");
29         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "warning");
30         
31         channel.basicQos(1);
32         
33         //定義一個消費者
34         Consumer consumer=new DefaultConsumer(channel){
35             //訊息到達 觸發這個方法
36             @Override
37             public void handleDelivery(String consumerTag, Envelope envelope,
38                     BasicProperties properties, byte[] body) throws IOException {
39              
40                 String msg=new String(body,"utf-8");
41                 System.out.println("[2] Recv msg:"+msg);
42                 
43                 try {
44                     Thread.sleep(2000);
45                 } catch (InterruptedException e) {
46                     e.printStackTrace();
47                 }finally{
48                     System.out.println("[2] done ");
49                     channel.basicAck(envelope.getDeliveryTag(), false);
50                 }
51             }
52         };
53         
54         boolean autoAck=false;//自動應答 false
55         channel.basicConsume(QUEUE_NAME,autoAck , consumer);
56     }
57         
58 }

Topic模型

 1 public class Send {
 2  private final static String EXCHANGE_NAME = "test_exchange_topic";
 3  public static void main(String[] argv) throws Exception {
 4  // 獲取到連線以及mq通道
 5  Connection connection = ConnectionUtils.getConnection();
 6  Channel channel = connection.createChannel();
 7  // 宣告exchange
 8  channel.exchangeDeclare(EXCHANGE_NAME, "topic");
 9  // 訊息內容
10  String message = "id=1001";
11  channel.basicPublish(EXCHANGE_NAME, "item.delete", null, message.getBytes());
12  System.out.println(" [x] Sent '" + message + "'");
13  channel.close();
14  connection.close();
15  }
16 }

消費者

 1 public class Recv {
 2 private final static String QUEUE_NAME = "test_queue_topic_1";
 3 private final static String EXCHANGE_NAME = "test_exchange_topic";
 4 public static void main(String[] argv) throws Exception {
 5 // 獲取到連線以及mq通道
 6 Connection connection = ConnectionUtils.getConnection();
 7 final Channel channel = connection.createChannel();
 8 // 宣告佇列
 9 channel.queueDeclare(QUEUE_NAME, false, false, false, null);
10 // 繫結佇列到交換機
11 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update");
12 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");
13 // 同一時刻伺服器只會發一條訊息給消費者
14 channel.basicQos(1);
15 // 定義佇列的消費者
16 Consumer consumer = new DefaultConsumer(channel) {
17 // 訊息到達 觸發這個方法
18 @Override
19 public void handleDelivery(String consumerTag, Envelope envelope,
20 BasicProperties properties, byte[] body) throws IOException {
21 String msg = new String(body, "utf-8");
22 System.out.println("[2] Recv msg:" + msg);
23 try {
24 Thread.sleep(1000);
25 } catch (InterruptedException e) {
26 e.printStackTrace();
27 } finally {
28 System.out.println("[2] done ");
29 // 手動回執
30 channel.basicAck(envelope.getDeliveryTag(), false);
31 }
32 }
33 };
34 boolean autoAck = false;
35 channel.basicConsume(QUEUE_NAME, autoAck, consumer);
36 }
37 }

消費者2

 1 public class Recv {
 2 private final static String QUEUE_NAME = "test_queue_topic_1";
 3 private final static String EXCHANGE_NAME = "test_exchange_topic";
 4 public static void main(String[] argv) throws Exception {
 5 // 獲取到連線以及mq通道
 6 Connection connection = ConnectionUtils.getConnection();
 7 final Channel channel = connection.createChannel();
 8 // 宣告佇列
 9 channel.queueDeclare(QUEUE_NAME, false, false, false, null);
10 // 繫結佇列到交換機
11 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update");
12 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");
13 // 同一時刻伺服器只會發一條訊息給消費者
14 channel.basicQos(1);
15 // 定義佇列的消費者
16 Consumer consumer = new DefaultConsumer(channel) {
17 // 訊息到達 觸發這個方法
18 @Override
19 public void handleDelivery(String consumerTag, Envelope envelope,
20 BasicProperties properties, byte[] body) throws IOException {
21 String msg = new String(body, "utf-8");
22 System.out.println("[2] Recv msg:" + msg);
23 try {
24 Thread.sleep(1000);
25 } catch (InterruptedException e) {
26 e.printStackTrace();
27 } finally {
28 System.out.println("[2] done ");
29 // 手動回執
30 channel.basicAck(envelope.getDeliveryTag(), false);
31 }
32 }
33 };
34 boolean autoAck = false;
35 channel.basicConsume(QUEUE_NAME, autoAck, consumer);
36 }
37 }

Exchanges（轉發器|交換機）

轉發器一方面它接受生產者的訊息，另一方面向佇列推送訊息

Nameless exchange（匿名轉發）

之前我們對轉換器一無所知，卻可以將訊息傳送到佇列，那是可能是我們用了預設的轉發器，轉發器名為空字串""。之前我們釋出訊息的程式碼

Fanout Exchange

不處理路由鍵。你只需要將佇列繫結到交換機上。傳送訊息到交換機都會被轉發到與該交換機繫結的所有佇列

 

Direct Exchange

處理路由鍵。
需要將一個佇列繫結到交換機上，要求該訊息與一個特定的路由鍵完全匹配。這是一個完整的匹配。如果一個佇列繫結到該交換機上要求路由鍵 “dog”，則只有被標記為“dog”的訊息才被轉發，不會轉發 dog.puppy，也不會轉發dog.guard，只會轉發 dog。
                                                    

Topic Exchange

將路由鍵和某模式進行匹配。
此時佇列需要繫結要一個模式上。符號“#”匹配一個或多個詞，符號“*”匹配一個詞。因此“audit.#”能夠匹配到“audit.irs.corporate”，但是“audit.*” 只會匹配到“audit.irs”。

 

相關推薦

RabbitMQ 的路由模式 Topic模式

模型 生產者 1 package cn.wh; 2 3 import java.io.IOException; 4 import java.util.concurrent.TimeoutException; 5 6 import cn.util.RabbitMqConn

RabbitMQ廣播：topic模式

topic模式跟direct差不多，只是把type改一下就行。 direct是把固定的routing_key跟queue繫結，topic是把模糊的routing_key跟queue繫結   原理圖： 釋出者： ''' 釋出者publisher ''' import pika im

RabbitMQ 訊息佇列 - topic 模式分發訊息

推薦閱讀 https://blog.csdn.net/column/details/15500.html topic 模式 根據 Binding 指定的 RoutingKey, Exchange 對 key 進行模式匹配後投遞到相應的 Queue, 模式匹配時符號

RabbitMQ之路由模式RoutingKey模式

生產者傳送訊息到交換機並指定一個路由key，消費者佇列繫結到交換機時要制定路由key（key匹配就能接受訊息，key不匹配就不能接受訊息） 例如：我們可以把路由key設定為insert ，那麼消費者佇列key指定包含insert才可以接收訊息，消費者佇列key定義為update或者delete就不能接收

RabbitMQ三種Exchange模式(fanout,direct,topic)的效能比較

RabbitMQ中，所有生產者提交的訊息都由Exchange來接受，然後Exchange按照特定的策略轉發到Queue進行儲存 RabbitMQ提供了四種Exchange：fanout,direct,topic,header header模式在實際使用中較少，本文只對

RabbitMQ (六) : 訂閱者模式之路由模式 ( direct )

efault rect 事件 .get 我們 兩個 create 發送消息 hello 路由模式下,生產者發送消息時需要指定一個路由鍵(routingKey),交換機只會把消息轉發給包含該路由鍵的隊列 這裏,我們改變一下聲明交換機的方式. 我們通過管理後臺添加一個交換機

RabbitMQ (七) : 訂閱者模式之主體模式 ( topic )

client 生產 精確 enc connect 技術分享 包含 get hang 主體模式和路由模式很像 路由模式是精確匹配 主體模式是模糊匹配 依然先通過管理後臺添加一個交換機. 生產者 public class Producer {

RabbitMQ三種Exchange模式(fanout,direct,topic)介紹

簡介： RabbitMQ中，所有生產者提交的訊息都由Exchange來接受，然後Exchange按照特定的策略轉發到Queue進行儲存 RabbitMQ提供了四種Exchange：fanout,direct,topic,header header模式在實際

topic模式下的收發

bind topic callback param AR ive tde binding change 生產者： import pika import sys connection = pika.BlockingConnection(pika.ConnectionPar

配置RabbitMQ默認群集模式

分發 服務器 oca python lda java 節點 src RoCE RabbitMQ是什麽？ MQ（Msaaage Queue，消息隊列）是一種應用程序對應用程序的通信方式。應用程序通過讀寫出入隊列的消息（針對應用程序的數據）來通信，而無須專用鏈接來連接它們。消

RabbitMQ廣播：fanout模式

一、 訊息的廣播需要exchange：exchange是一個轉發器，其實把訊息發給RabbitMQ裡的exchange fanout: 所有bind到此exchange的queue都可以接收訊息，廣播 direct: 通過routingKey和exchange決定的那個唯一的queue可以接收訊息

RabbitMQ廣播：direct模式

一、 訊息的廣播需要exchange：exchange是一個轉發器，其實把訊息發給RabbitMQ裡的exchange fanout: 所有bind到此exchange的queue都可以接收訊息，廣播 direct: 通過routingKey和exchange決定的那個唯一的queue可以接收訊息 t

activemq實戰整合Spring之Topic模式

這一節我們看一下Topic模式下的訊息釋出是如何處理的。 applicationContext-ActiveMQ.xml配置： <?xml version="1.0" encoding="UTF-8"?> <beans xmlns="htt

RabbitMQ Java 訂閱釋出模式

所有訂閱者都可以獲取到來自交換機轉發的任務訊息 連線到Rabbit，我們都需要一個新的空的佇列，消費者與Rabbit斷開，消費者所接收的那個佇列應該被自動刪除 pom.xml   <project xmlns="http://maven.apache.

設計模式-builder模式（以微信訊息的路由處理為例）

今天要講一個簡單的模式–builder模式。 你可能會覺得,builder模式有什麼好講的？本來我也這樣覺得，但當我有幸拜讀某位大神通過builder模式寫了一個開發工具包的初始化操作，嗯程式碼的樣子很叼 這位大神是誰呢？ 就是這位仁兄啦，是碼雲上一個很火的微信開發工具包的貢獻者之一。

RabbitMQ 訊息佇列 - fanout 模式分發訊息

推薦閱讀 https://blog.csdn.net/column/details/15500.html fanout 模式 將同一個 message 傳送到所有同該 Exchange 繫結的 queue, 只要 RoutingKey 是一樣, 這條訊息都會被投遞

RabbitMQ 訊息佇列 - direct 模式分發訊息

推薦閱讀 https://blog.csdn.net/column/details/15500.html direct 模式 根據 Binding 指定的 Routing Key, 將符合Key的訊息傳送到 Binding 的 Queue p_direc

rabbitmq7-topic模式

一、簡介 如果你對rabbitmq6-direct路由模式比較熟悉的話，把這個簡介看完然後跳過，如果不熟的話，就把下面的程式碼看一看，有興趣的敲一敲就可以了。本來不想寫這一節的，但是為了體系的完整性，所以還是把這一節給補出來。 在rabbitmq2-這可能是rabbitmq最全的概覽

5.rabbitmq--萬用字元模式Topics

rabbitmq--萬用字元模式Topics topic模式也稱為主題模式，其實他相對於routing模式最大的好處就是他多了一種匹配模式的路由，怎麼理解匹配呢，其實就相當於我們之前正則的.*這種，不過他的匹配機制可能不是這種（其實除了匹配規則外，他的作用就和routing模式一樣 ），而他的工作

RabbitMQ系列之三 RabbitMQ幾種典型模式

本文詳細介紹簡單模式Simple、工作模式Work、釋出訂閱模式Publish/Subscribe、路由模式Routing、萬用字元模式Topics、遠端呼叫模式RPC(暫不對該佇列模式進行詳解) 模式1：簡單模式（Simple / HelloWorld 單生產單消費） 簡單的傳送與