2. 程式人生 > >【RabbitMQ】三種Exchange模式——訂閱、路由、萬用字元模式

【RabbitMQ】三種Exchange模式——訂閱、路由、萬用字元模式

阿新 • • 發佈：2019-01-06

   前兩篇部落格介紹了兩種佇列模式，這篇部落格介紹訂閱、路由和萬用字元模式，之所以放在一起介紹，是因為這三種模式都是用了Exchange交換機，訊息沒有直接傳送到佇列，而是傳送到了交換機，經過佇列繫結交換機到達佇列。

一、訂閱模式(Fanout Exchange)：

   一個生產者，多個消費者，每一個消費者都有自己的一個佇列，生產者沒有將訊息直接傳送到佇列，而是傳送到了交換機，每個佇列繫結交換機，生產者傳送的訊息經過交換機，到達佇列，實現一個訊息被多個消費者獲取的目的。需要注意的是，如果將訊息傳送到一個沒有佇列繫結的exchange上面，那麼該訊息將會丟失，這是因為在rabbitMQ中exchange不具備儲存訊息的能力，只有佇列具備儲存訊息的能力。

      

         

示例程式碼：

生產者：

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();

        // 宣告exchange
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

        // 訊息內容
        String message = "商品已經新增，id = 1000";
        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
        
        System.out.println(" [x] Sent '" + message + "'");

        channel.close();
        connection.close();
    }
}
消費者1： 
public class Recv {

    private final static String QUEUE_NAME = "test_queue_fanout_1";

    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();

        // 宣告佇列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 繫結佇列到交換機
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

        // 同一時刻伺服器只會發一條訊息給消費者
        channel.basicQos(1);

        // 定義佇列的消費者
        QueueingConsumer consumer = new QueueingConsumer(channel);
        // 監聽佇列，手動返回完成
        channel.basicConsume(QUEUE_NAME, true, consumer);

        // 獲取訊息
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" 前臺系統： '" + message + "'");
            Thread.sleep(10);

            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}

  消費者2的程式碼和消費者1的程式碼大致相同，只是佇列的名稱不一樣，這樣兩個消費者有自己的佇列，都可以接收到生產者傳送的訊息


  但是如果生產者有新增商品，修改商品，刪除商品的訊息，消費者包快前臺系統和搜尋系統，要求前臺系統接收修改和刪除商品的訊息，搜尋系統接收新增商品、修改商品和刪除商品的訊息。所以使用這種訂閱模式實現商品資料的同步並不合理。因此我們介紹下一種模式：路由模式。

二、路由模式(Direct Exchange)

  這種模式添加了一個路由鍵，生產者釋出訊息的時候新增路由鍵，消費者繫結佇列到交換機時新增鍵值，這樣就可以接收到需要接收的訊息。

       


示例程式碼：

生產者： 
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();

        // 宣告exchange
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");

        // 訊息內容
        String message = "刪除商品， id = 1001";
        channel.basicPublish(EXCHANGE_NAME, "delete", null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");

        channel.close();
        connection.close();
    }
}
消費者1：接收更新和刪除訊息 
public class Recv {

    private final static String QUEUE_NAME = "test_queue_direct_1";

    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");

        // 同一時刻伺服器只會發一條訊息給消費者
        channel.basicQos(1);

        // 定義佇列的消費者
        QueueingConsumer consumer = new QueueingConsumer(channel);
        // 監聽佇列，手動返回完成
        channel.basicConsume(QUEUE_NAME, false, consumer);

        // 獲取訊息
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" 前臺系統： '" + message + "'");
            Thread.sleep(10);

            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}
消費者2：接收insert，update，delete的訊息 
public class Recv2 {

    private final static String QUEUE_NAME = "test_queue_direct_2";

    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");

        // 同一時刻伺服器只會發一條訊息給消費者
        channel.basicQos(1);

        // 定義佇列的消費者
        QueueingConsumer consumer = new QueueingConsumer(channel);
        // 監聽佇列，手動返回完成
        channel.basicConsume(QUEUE_NAME, false, consumer);

        // 獲取訊息
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" 搜尋系統： '" + message + "'");
            Thread.sleep(10);

            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}
  如果生產者釋出了insert訊息，那麼消費者2可以收到，消費者 1收不到，如果釋出了update或者delete訊息，兩個消費者都可以收到。如果釋出ABC訊息兩個消費者都收不到，因為沒有繫結這個鍵值。這種模式基本滿足了我們的需求，但是還不夠靈活，下面介紹另外一個模式。

三、萬用字元模式（Topic Exchange）

   基本思想和路由模式是一樣的，只不過路由鍵支援模糊匹配，符號“#”匹配一個或多個詞，符號“*”匹配不多不少一個詞

       

示例程式碼：

生產者：

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();

        // 宣告exchange
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");

        // 訊息內容
        String message = "刪除商品，id = 1001";
        channel.basicPublish(EXCHANGE_NAME, "item.delete", null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");

        channel.close();
        connection.close();
    }
}

消費者1： 
public class Recv {

    private final static String QUEUE_NAME = "test_queue_topic_1";

    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();

        // 宣告佇列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 繫結佇列到交換機
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete");

        // 同一時刻伺服器只會發一條訊息給消費者
        channel.basicQos(1);

        // 定義佇列的消費者
        QueueingConsumer consumer = new QueueingConsumer(channel);
        // 監聽佇列，手動返回完成
        channel.basicConsume(QUEUE_NAME, false, consumer);

        // 獲取訊息
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" 前臺系統： '" + message + "'");
            Thread.sleep(10);

            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}


消費者2：

public class Recv2 {

    private final static String QUEUE_NAME = "test_queue_topic_2";

    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();

        // 宣告佇列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 繫結佇列到交換機
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.#");

        // 同一時刻伺服器只會發一條訊息給消費者
        channel.basicQos(1);

        // 定義佇列的消費者
        QueueingConsumer consumer = new QueueingConsumer(channel);
        // 監聽佇列，手動返回完成
        channel.basicConsume(QUEUE_NAME, false, consumer);

        // 獲取訊息
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" 搜尋系統： '" + message + "'");
            Thread.sleep(10);

            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}

  消費者1是按需索取，並沒有使用萬用字元模式，而是用的完全匹配，消費者2使用萬用字元模式，這樣以item.開頭的訊息都會全部接收。

小結：

  1.與簡單模式和work模式對比，前面兩種同一個訊息只能被一個消費者獲取，而今天的這三種模式，可以實現一個訊息被多個消費者 獲取。

  2.fanout這種模式沒有加入路由器，佇列與exchange繫結後，就會接收到所有的訊息，其餘兩種增加了路由鍵，並且第三種增加萬用字元，更加便利。

相關推薦

【RabbitMQ】三種Exchange模式——訂閱、路由、萬用字元模式

   前兩篇部落格介紹了兩種佇列模式，這篇部落格介紹訂閱、路由和萬用字元模式，之所以放在一起介紹，是因為這三種模式都是用了Exchange交換機，訊息沒有直接傳送到佇列，而是傳送到了交換機，經過佇列繫結交換機到達佇列。 一、訂閱模式(Fanout Exchange)：  

【Android】三種工廠模式簡析

簡單介紹三種工廠模式：簡單工廠模式，工廠方法模式，抽象工廠模式 1.簡單工廠模式 這是一個小工廠，什麼都由自己生產，別人要什麼就呼叫自己的相應工具去生產 具備三個特徵：具體工廠A，具體生產工具B，抽象生產物件C 就是 A呼叫工具B去生產C

【RabbitMQ】三種類型交換器 Fanout,Direct,Topic（轉）

出處：https://blog.csdn.net/fxq8866/article/details/62049393 RabbitMQ伺服器會根據路由鍵將訊息從交換器路由到佇列中，如何處理投遞到多個佇列的情況？這裡不同型別的交換器起到了重要的作用。分別是fanout,direct，topic，每一種型別實現了

【C#】三種轉換字元法、三元表示式、常量列舉結構和陣列

一、三種方法轉換字串變成數字 1.1.Parse的使用 這兩個使用結果一樣，當用戶輸入的數字轉換成我們想要的數字型別。 可以使用     Convert.ToInt32(""); 還可以使用  int.Parse("") 1.2.int.TryParse的使用

thinkphp5入門【一】三種安裝方法

ThinkPHP V5.0是一個為API開發而設計的高效能框架——是一個顛覆和重構版本，採用全新的架構思想，引入了很多的PHP新特性，優化了核心，減少了依賴，實現了真正的惰性載入，支援composer，並針對API開發做了大量的優化。 ThinkPHP5是一個全新的里程碑版本，包括路由、

【轉】三種快速排序演算法以及快速排序的優化

一.  快速排序的基本思想 快速排序使用分治的思想，通過一趟排序將待排序列分割成兩部分，其中一部分記錄的關鍵字均比另一部分記錄的關鍵字小。之後分別對這兩部分記錄繼續進行排序，以達到整個序列有序的目的。 二.  快速排序的三個步驟 1) 選擇基準：在待排序列中，按照某種方式挑出一個元素，作為 “基準”（p

【RabbitMQ】三種類型交換器 Fanout,Direct,Topic

  RabbitMQ伺服器會根據路由鍵將訊息從交換器路由到佇列中，如何處理投遞到多個佇列的情況？這裡不同型別的交換器起到了重要的作用。分別是fanout,direct，topic，每一種型別實現了不同

RabbitMQ的三種Exchange，資料持久化,非持久化例項

由於最近專案需要用到Rabbitmq的資料持久化技術，利用空閒時間，分別對Rabbitmq的三種常用的Exchange（direct、fanout、topic）寫了個測試例項，僅供初學者參考學習，還望各路大神勿吐槽。 開發之前需要引用包：最好是3.4.0以下版本，因為以上版

【leetcode】迴文連結串列（Palindrome Linked List）【python】三種方法

題目連結 時間複雜度O(N),空間複雜度O(N) class ListNode: def __init__(self, x): self.val = x se

【C++】三種呼叫類的複製建構函式的情況

用類的一個物件初始化同類的另一個物件時。 某函式的返回值是類的物件，呼叫該函式時。 某函式的形參是類的物件，呼叫該函式時。 ※注意區分“初始化”和“賦值”： ClassName c2 = c1;    （初始化語句） ClassName c1 , c2;

【Java】三種方式建立多執行緒

通過整合Thread類實現多執行緒 package TreadLearning; /** * 建立執行緒方法1 * 利用繼承Thread類+重寫run方法 * new物件然後呼叫start（）方法（不能直接呼叫run方法） * @author 袁盛桐 * */

【Python】三種遍歷檔案目錄的方法

#coding:utf-8 # 方法1：遞迴遍歷目錄 import os def visitDir(path): li = os.listdir(path) for p in li

【NLP】三種知識圖譜

三種知識圖譜 知識圖譜本質上是一種語義網路。其結點代表實體（entity）或者概念（concept），邊代表實體/概念之間的各種語義關係。Knowledge Graph , Freebase , Wikidata 是目前最常見的三種知識圖譜。 Knowledge Gr

【VMware】三種網路連線的區別

1、概述 　　大家在安裝完虛擬機器後，預設安裝了兩個虛擬網絡卡，VMnet1和VMnet8，如下圖。其中VMnet1是host網絡卡，用於host方式連線網路的。VMnet8是NAT網絡卡，用於NAT方式連線網路的。它們的IP地址是隨機生成的。 　　VMware提

【RabbitMQ】——5種佇列

應用RabbitMQ，我們可以根據需求選擇5種佇列之一。 一、簡單佇列 P：訊息的生產者 C：訊息的消費者 紅色：佇列 簡單佇列的生產者和消費者關係一對一 但有時我們的需求，需要一個生產者，對應多個消費者，那就可以採用第二種模式 二、Wor

【轉】三種方法讓你在I2C通信中同時和多個從機通信

efi 抖動 萬能 stop ans ole println tran 正常 ref：http://tieba.baidu.com/p/3769008030對於不同地址的模塊就不用多說了，直接分別對其地址進行通信即可。那麽若拿到相同地址的模塊，或者直接是相同的多個模塊怎麽辦

【PLC】三菱PLC MITSUBISHI ELECTRIC FX3U-16M、FX3U-46M 型號PLC Monitor Utility控制元件通訊設定

配置 第一步、選擇通訊方式 – 串列埠通訊；選擇串列埠號 – COM3(示例)；設定超時時間。 第二部、選擇CPU型別 – FX3UC;波特率 – 115200；選擇控制方式-DTR and RTS。 第三步、完成配置，測試連線。  

RabbitMQ之Topics（萬用字元模式）

說明：此模式實在路由key模式的基礎上，使用了萬用字元來管理消費者接收訊息。生產者P傳送訊息到交換機X，type=topic，交換機根據繫結佇列的routing key的值進行萬用字元匹配； 符號#：匹配一個或者多個詞lazy.# 可以匹配lazy.irs或者lazy.irs.cor符號*：只能匹配一個詞l

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

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

RabbitMQ的萬用字元模式Demo

案例: 根據使用者的通知設定去通知使用者，設定接收Email的使用者只接收Email，設定接收sms的使用者只接收sms，設定兩種 通知型別都接收的則兩種通知都有效。 生產者： package com.xuecheng.rabbitmq.producer; import com.rab