【譯】RabbitMQ系列（五） - 主題模式

pingan8787 發(fā)布于2019-08-16 16:07 / 1155人閱讀

摘要：主題模式在上一章我們改進了我們的日志系統(tǒng)，如果使用我們只能簡單進行廣播，而使用則允許消費者可以進行一定程度的選擇。為的會同時發(fā)布到這兩個。當為時，會接收所有的。當中沒有使用通配符和時，的行為和一致。

主題模式

在上一章我們改進了我們的日志系統(tǒng)，如果使用fanout我們只能簡單進行廣播，而使用direct則允許消費者可以進行一定程度的選擇。但是direct還是有其局限性，其路由不支持多個條件。

在我們的日志系統(tǒng)中，消費者程序可能不止是基于日志的severity，同時也想基于發(fā)送日志的源系統(tǒng)。你可能知道linux的syslog工具，它就是同時基于severity(info/warn/crit...)和功能(auth/cron/kern...).

這就提供了很大的靈活性-我們想接收來自cron的嚴重錯誤日志和kern的所有日志。

下面我們就使用更復(fù)雜的topic來改進我們的日志系統(tǒng)。

Topic exchange

發(fā)送到topic類型exchange的message不可以具有模糊的routing_key,它必須具有以冒號分割的詞。就像"stock.usd.nyse","nyse.vmw","quick.orange.rabbit"等，限制長度255字節(jié)。

binding key也采用相似的形勢。topic exchange的邏輯和direct相似，通過比較message的routing key和bind的binding key，來匹配轉(zhuǎn)發(fā)的queue。但是topic的binding支持通配符：

” * “表示任何一個詞

” # “ 表示0或1個詞

通過上面圖示的場景來解釋會比較好理解。

例子中我們將發(fā)送描述動物的message。message會攜帶routing key（包含三個詞），第一個詞表示speed，第二個表示color，第三個表示species"..".

創(chuàng)建了三個綁定：Q1的binding key是”*.orange.*" Q2的binding key是“*.*.rabbit”和 "lazy.#".

以文字表述便是：

Q1 關(guān)心所有橘色的動物

Q2 關(guān)心所有的rabbit和所有的lazy動物

routing key為“quick.orange.rabbit"的message會同時發(fā)布到這兩個queue。
routing key為"lazy.orange.elephant"的message會同時發(fā)布到這兩個queue。
routing key為”quick.orange.fox“只會發(fā)布到第一個queue.
routing key為”lazy.brown.fox"的message只會發(fā)布到第二個queue.
routing key為"lazy.pink.rabbit"的message雖然滿足Q2的兩個條件，但也只會發(fā)布到Q2一次。
routing key為"quick.brown.fox"的message沒有任何匹配，就會被丟失。

如果我們發(fā)送的message只有一個word或者多余三個word，如"orange"或者"quick.orange.male.rabbit"會發(fā)生什么呢？這些message不會匹配任何binding key，均會被丟棄掉。

另外"lazy.orange.male.rabbit"雖然具有四個詞，但是會匹配最后的binding key，而被發(fā)送到第二個queue。

Topic exhange非常強大，同時可以模仿其他兩種類型的exchange。當binding key為 # 時，queue會接收所有的message。當binding key中沒有使用通配符(* 和 #)時，topic的行為和direct一致。

開始執(zhí)行

我們將在日志系統(tǒng)中使用topic exchange。我們的routding key采用兩個詞 ".".
EmitLogTopic.java的代碼如下：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class EmitLogTopic {

  private static final String EXCHANGE_NAME = "topic_logs";

  public static void main(String[] argv) throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    try (Connection connection = factory.newConnection();
         Channel channel = connection.createChannel()) {

        channel.exchangeDeclare(EXCHANGE_NAME, "topic");

        String routingKey = getRouting(argv);
        String message = getMessage(argv);

        channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
        System.out.println(" [x] Sent "" + routingKey + "":"" + message + """);
    }
  }
  //..
}

ReceiveLogsTopic.java的代碼如下:

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

public class ReceiveLogsTopic {

  private static final String EXCHANGE_NAME = "topic_logs";

  public static void main(String[] argv) throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();

    channel.exchangeDeclare(EXCHANGE_NAME, "topic");
    String queueName = channel.queueDeclare().getQueue();

    if (argv.length < 1) {
        System.err.println("Usage: ReceiveLogsTopic [binding_key]...");
        System.exit(1);
    }

    for (String bindingKey : argv) {
        channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
    }

    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

    DeliverCallback deliverCallback = (consumerTag, delivery) -> {
        String message = new String(delivery.getBody(), "UTF-8");
        System.out.println(" [x] Received "" +
            delivery.getEnvelope().getRoutingKey() + "":"" + message + """);
    };
    channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
  }
}

編譯

javac -cp $CP ReceiveLogsTopic.java EmitLogTopic.java

接收所有日志

java -cp $CP ReceiveLogsTopic "#"

接收功能"kern"的日志

java -cp $CP ReceiveLogsTopic "kern.*"

接收嚴重級別日志

java -cp $CP ReceiveLogsTopic "*.critical"

接收者使用兩個綁定條件

java -cp $CP ReceiveLogsTopic "kern.*" "*.critical"

發(fā)送日志message

java -cp $CP EmitLogTopic "kern.critical" "A critical kernal error"

GPU云服務(wù)器云服務(wù)器 Java設(shè)計模式之（五）——代理模式 RabbitMQ rabbitmq 事務(wù)消息 rabbitmq 消息順序

文章版權(quán)歸作者所有，未經(jīng)允許請勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。

轉(zhuǎn)載請注明本文地址：http://m.specialneedsforspecialkids.com/yun/73916.html

發(fā)表評論

登陸后可評論

0條評論

pingan8787

男|高級講師

我要關(guān)注我要私信

TA的文章

查詢tensorflow版本

閱讀 698·2023-04-25 22:50
記錄不存在則插入，存在則更新 → MySQL 的實現(xiàn)方式有哪些？

閱讀 1531·2021-10-08 10:05
阿里云存儲，讓更多的創(chuàng)新在身邊發(fā)生

閱讀 987·2021-09-30 09:47
自動化測試系列（三）|UI測試

閱讀 1921·2021-09-28 09:35
C++入門—namespace的使用傻傻分不清楚&C++中函數(shù)的參數(shù)也可以配備胎&a

閱讀 825·2021-09-26 09:55
大數(shù)求余C++（超級牛逼）

閱讀 3416·2021-09-10 10:51
【C語言進階】字符串&內(nèi)存函數(shù)

閱讀 3431·2021-09-02 15:15
80VPS：日本/香港CN2服務(wù)器首月420元(續(xù)費600元),E5/16G/1TB/20M帶寬

閱讀 3297·2021-08-05 09:57

国产xxxx99真实实拍_久久不雅视频_高清韩国a级特黄毛片_嗯老师别我我受不了了小说

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務(wù)器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！

【譯】RabbitMQ系列（五） - 主題模式

相關(guān)文章

**【譯】RabbitMQ系列（四） - 路由模式**

**白話RabbitMQ(五): 主題路由器(Topic Exchange)**

**【譯】RabbitMQ系列（二）-Work模式**

**【譯】RabbitMQ系列（三） - 發(fā)布/訂閱模式**

**【譯】RabbitMQ系列（六）-RPC模式**

發(fā)表評論

0條評論

pingan8787

男|高級講師

TA的文章

查詢tensorflow版本

記錄不存在則插入，存在則更新 → MySQL 的實現(xiàn)方式有哪些？

阿里云存儲，讓更多的創(chuàng)新在身邊發(fā)生

自動化測試系列（三）|UI測試

C++入門—namespace的使用傻傻分不清楚&C++中函數(shù)的參數(shù)也可以配備胎&a

大數(shù)求余C++（超級牛逼）

【C語言進階】字符串&內(nèi)存函數(shù)

80VPS：日本/香港CN2服務(wù)器首月420元(續(xù)費600元),E5/16G/1TB/20M帶寬

最新活動

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務(wù)器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！

【譯】RabbitMQ系列（五） - 主題模式

相關(guān)文章

發(fā)表評論

0條評論

男|高級講師

TA的文章

最新活動

上云采購季！| 2核2G4M爆款云服務(wù)器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！