白話RabbitMQ(四): 建立路由

CoderStudy 發(fā)布于2019-08-15 13:29 / 2144人閱讀

摘要：可以參考源碼，項(xiàng)目支持網(wǎng)站，最新文章或?qū)崿F(xiàn)會(huì)更新在上面前言在訂閱發(fā)布中我們建立了一個(gè)簡單的日志系統(tǒng)，從而將消息廣播給一些消費(fèi)者。因此，發(fā)送到路由鍵的消息會(huì)發(fā)送給隊(duì)列，發(fā)送到路由鍵或者的消息會(huì)發(fā)送給，其它的消息將被丟棄。

推廣

RabbitMQ專題講座

https://segmentfault.com/l/15...

CoolMQ開源項(xiàng)目

我們利用消息隊(duì)列實(shí)現(xiàn)了分布式事務(wù)的最終一致性解決方案，請大家圍觀?？梢詤⒖荚创a：https://github.com/vvsuperman…，項(xiàng)目支持網(wǎng)站: http://rabbitmq.org.cn，最新文章或?qū)崿F(xiàn)會(huì)更新在上面

前言

在訂閱/發(fā)布中我們建立了一個(gè)簡單的日志系統(tǒng)，從而將log消息廣播給一些消費(fèi)者。這章我們會(huì)在此基礎(chǔ)上加入一些新的特性-我們將有針對性的進(jìn)行消息分發(fā)，比如，只把錯(cuò)誤(error)消息保存到磁盤，與此同時(shí)，打印出所有的消息。

綁定

我們在前面的例子中，綁定是這么來做的

channel.queueBind(queueName, EXCHANGE_NAME, "");

綁定是建立交換機(jī)和隊(duì)列之間的一種聯(lián)系：隊(duì)列會(huì)接受交換機(jī)中的消息。綁定可以用一個(gè)路由鍵來指明，為了與basic_publish區(qū)分開，我們稱之為綁定鍵(binding key):

channel.queueBind(queueName, EXCHANGE_NAME, "black");

綁定鍵跟路由器類型也有關(guān)系，我們之前用的廣播路由器，會(huì)忽略掉這個(gè)值

直達(dá)交換機(jī)(Direct Exchange)

之前我們用的是廣播交換機(jī)，會(huì)將消息發(fā)送給所有的消費(fèi)者。這里我們希望通過log的嚴(yán)重程度進(jìn)行過濾，例如只有嚴(yán)重的錯(cuò)誤才會(huì)寫入到磁盤，而warn和info消息就不用了，以此來節(jié)省磁盤空間

而廣播交換機(jī)沒法滿足這個(gè)需求-它只是無腦的發(fā)送消息。所以我們會(huì)使用直達(dá)交換機(jī)(Direct Exchange)- 消息會(huì)通過所綁定的鍵來發(fā)送給對應(yīng)的隊(duì)列，可以看如下這幅圖

如上圖所示，直達(dá)交換機(jī)X綁定了兩個(gè)隊(duì)列，C1是通過orange來綁定，而C2是通過black和green綁定。因此，發(fā)送到路由鍵orange的消息會(huì)發(fā)送給隊(duì)列Q1，發(fā)送到路由鍵black或者green的消息會(huì)發(fā)送給Q2，其它的消息將被丟棄。

多項(xiàng)綁定

當(dāng)然，多個(gè)隊(duì)列綁定到一個(gè)鍵上也是合法的，在這種情況下，直達(dá)交換機(jī)將會(huì)將消息發(fā)送給所有的隊(duì)列，就像廣播交換機(jī)一樣，如上圖所示，一個(gè)鍵為black的消息將會(huì)同時(shí)被發(fā)送給C1和C2.

我們首先需要?jiǎng)?chuàng)建一個(gè)直達(dá)路由器

channel.exchangeDeclare(EXCHANGE_NAME, "direct");

并發(fā)送消息到這個(gè)路由器

channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());

上面我們是發(fā)送給"severity",簡單起見，假設(shè)有下列幾種日志類型"severity" ,"info", "warning", "error".

訂閱消息(Subscribing)

接受消息跟之前一樣，但有一點(diǎn)不同，我們提供了一個(gè)binding key，

String queueName = channel.queueDeclare().getQueue();

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

整合

將上面的所有代碼整合到一起

EmitLogDirect.java

import com.rabbitmq.client.*;

import java.io.IOException;

public class EmitLogDirect {

    private static final String EXCHANGE_NAME = "direct_logs";

    public static void main(String[] argv)
                  throws java.io.IOException {

        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare(EXCHANGE_NAME, "direct");

        String severity = getSeverity(argv);
        String message = getMessage(argv);

        channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
        System.out.println(" [x] Sent "" + severity + "":"" + message + """);

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

ReceiveLogsDirect.java

import com.rabbitmq.client.*;

import java.io.IOException;

public class ReceiveLogsDirect {

  private static final String EXCHANGE_NAME = "direct_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, "direct");
    String queueName = channel.queueDeclare().getQueue();

    if (argv.length < 1){
      System.err.println("Usage: ReceiveLogsDirect [info] [warning] [error]");
      System.exit(1);
    }

    for(String severity : argv){
      channel.queueBind(queueName, EXCHANGE_NAME, severity);
    }
    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

    Consumer consumer = new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String consumerTag, Envelope envelope,
                                 AMQP.BasicProperties properties, byte[] body) throws IOException {
        String message = new String(body, "UTF-8");
        System.out.println(" [x] Received "" + envelope.getRoutingKey() + "":"" + message + """);
      }
    };
    channel.basicConsume(queueName, true, consumer);
  }
}

編譯

javac -cp $CP ReceiveLogsDirect.java EmitLogDirect.java

只保存warning和error的消息到磁盤上

java -cp $CP ReceiveLogsDirect warning error > logs_from_rabbit.log

將所有的消息打印到頻幕上

java -cp $CP ReceiveLogsDirect info warning error
# => [*] Waiting for logs. To exit press CTRL+C

最后，發(fā)送error消息

java -cp $CP EmitLogDirect error "Run. Run. Or it will explode."
# => [x] Sent "error":"Run. Run. Or it will explode."

好了，這一章就到這兒，下一章我們將講述如何基于特定模式進(jìn)行監(jiān)聽

云服務(wù)器 GPU云服務(wù)器小米路由器如何建立云服務(wù)器嗎白話白話云計(jì)算白話區(qū)塊鏈

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

轉(zhuǎn)載請注明本文地址：http://systransis.cn/yun/68120.html

上一篇：android開發(fā)新手---自定義標(biāo)題欄（踩坑?。?/a>

下一篇：Dubbo 新編程模型之外部化配置

發(fā)表評論

登陸后可評論

0條評論

CoderStudy

男|高級講師

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

TA的文章

（補(bǔ)貨）HostYun：洛杉磯Ceranetworks機(jī)房AMD YES系列VPS，美國原生IP，去

閱讀 1276·2021-10-18 13:32
ABD — android debug bridge 簡略介紹

閱讀 2355·2021-09-24 09:47
小米回應(yīng)立陶宛：遵守歐盟通用數(shù)據(jù)保護(hù)條例尊重用戶合法權(quán)益

閱讀 1336·2021-09-23 11:22
【CSS基礎(chǔ)】Flex彈性布局

閱讀 2473·2019-08-30 14:06
HTML/CSS基礎(chǔ)知識(shí)總結(jié)

閱讀 579·2019-08-30 12:48
css權(quán)重與常見布局（1）

閱讀 2010·2019-08-30 11:03
Vue.js 渲染簡寫樣式存在的問題

閱讀 546·2019-08-29 17:09
前端小知識(shí)--為什么你寫的height:100%不起作用？

閱讀 2473·2019-08-29 14:10

成人国产在线小视频_日韩寡妇人妻调教在线播放_色成人www永久在线观看_2018国产精品久久_亚洲欧美高清在线30p_亚洲少妇综合一区_黄色在线播放国产_亚洲另类技巧小说校园_国产主播xx日韩_a级毛片在线免费

資訊專欄INFORMATION COLUMN

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

白話RabbitMQ(四): 建立路由

RabbitMQ專題講座

CoolMQ開源項(xiàng)目

相關(guān)文章

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

**白話RabbitMQ(三):發(fā)布/訂閱**

**白話rabbitmq(一): HelloWorld**

**白話RabbitMQ(六): RPC**

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

發(fā)表評論

0條評論

CoderStudy

男|高級講師

TA的文章

（補(bǔ)貨）HostYun：洛杉磯Ceranetworks機(jī)房AMD YES系列VPS，美國原生IP，去

ABD — android debug bridge 簡略介紹

小米回應(yīng)立陶宛：遵守歐盟通用數(shù)據(jù)保護(hù)條例尊重用戶合法權(quán)益

【CSS基礎(chǔ)】Flex彈性布局

HTML/CSS基礎(chǔ)知識(shí)總結(jié)

css權(quán)重與常見布局（1）

Vue.js 渲染簡寫樣式存在的問題

前端小知識(shí)--為什么你寫的height:100%不起作用？

最新活動(dòng)

資訊專欄INFORMATION COLUMN

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

白話RabbitMQ(四): 建立路由

RabbitMQ專題講座

CoolMQ開源項(xiàng)目

相關(guān)文章

發(fā)表評論

0條評論

男|高級講師

TA的文章

最新活動(dòng)

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