2. 程式人生 > >訊息中介軟體——RabbitMQ（八）高階特性全在這裡!（下）

訊息中介軟體——RabbitMQ（八）高階特性全在這裡!（下）

阿新 • • 發佈：2019-08-28

前言

上一篇訊息中介軟體——RabbitMQ（七）高階特性全在這裡!（上）中我們介紹了訊息如何保障100%的投遞成功？,冪等性概念詳解,在海量訂單產生的業務高峰期，如何避免訊息的重複消費的問題？,Confirm確認訊息、Return返回訊息。這篇我們來介紹下下面內容。

  • 自定義消費者
  • 訊息的限流（防止佔用記憶體過多，節點宕機）
  • 訊息的ACK與重回佇列
  • TTL訊息
  • 死信佇列

1. 自定義消費者

1.1 消費端自定義監聽

我們一般就在程式碼中編寫while迴圈，進行consumer.nextDelivery方法進行獲取下一條訊息，然後進行消費處理！

但是這種輪訓的方式肯定是不好的，程式碼也比較low。

  • 我們使用自定義的Consumer更加的方便，解耦性更加的強，也是在實際工作中最常見的使用方式！

1.2 程式碼演示

1.2.1 生產者


/**
 * 
* @ClassName: Producer 
* @Description: 生產者
* @author Coder程式設計
* @date2019年7月30日 下午23:15:51 
*
 */
public class Producer {

    
    public static void main(String[] args) throws Exception {
        
        //1 建立ConnectionFactory
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        
        String exchange = "test_consumer_exchange";
        String routingKey = "consumer.save";
        
        String msg = "Hello RabbitMQ Consumer Message";
        
        for(int i =0; i<5; i ++){
            channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());
        }
        
    }
}

1.2.2 消費者


/**
 * 
* @ClassName: Consumer 
* @Description: 消費者
* @author Coder程式設計
* @date2019年7月30日 下午23:13:51 
*
 */
public class Consumer {

    
    public static void main(String[] args) throws Exception {
        
        
        // 建立ConnectionFactory
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        
        
        String exchangeName = "test_consumer_exchange";
        String routingKey = "consumer.#";
        String queueName = "test_consumer_queue";
        
        channel.exchangeDeclare(exchangeName, "topic", true, false, null);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);
        
        //實現自己的MyConsumer()
        channel.basicConsume(queueName, true, new MyConsumer(channel));
    }
}

1.2.3 自定義類：MyConsumer

/**
 * 
* @ClassName: MyConsumer 
* @Description: TODO
* @author Coder程式設計
* @date 2019年7月30日 下午23:11:55 
*
 */
public class MyConsumer extends DefaultConsumer {


    public MyConsumer(Channel channel) {
        super(channel);
    }

    //根據需求，重寫自己需要的方法。
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        System.err.println("-----------consume message----------");
        //消費標籤
        System.err.println("consumerTag: " + consumerTag);
        //這個物件包含許多關鍵資訊
        System.err.println("envelope: " + envelope);
        System.err.println("properties: " + properties);
        System.err.println("body: " + new String(body));
    }


}

1.3 列印結果

2. 消費端限流

2.1 什麼是消費端的限流？

  • 假設一個場景，首先，我們Rabbitmq伺服器有上萬條未處理的訊息，我們隨便開啟一個消費者客戶端，會出現下面情況：
  • 巨量的訊息瞬間全部推送過來，但是我們單個客戶端無法同時處理這麼多資料！這個時候很容易導致伺服器崩潰，出現故障。

為什麼不在生產端進行限流呢？

因為在高併發的情況下，客戶量就是非常大，所以很難在生產端做限制。因此我們可以用MQ在消費端做限流。

  • RabbitMQ提供了一種qos(服務質量保證)功能，即在非自動確認訊息的前提下，如果一定數目的訊息（通過基於consume或者channel設定Qos的值）未被確認前，不進行消費新的訊息。
    在限流的情況下，千萬不要設定自動簽收，要設定為手動簽收。
  • void BasicQos(uint prfetchSize,ushort prefetchCount,bool global);

引數解釋：
prefetchSize:0
prefetchCount:會告訴RabbitMQ不要同時給一個消費者推送多於N個訊息，即一旦有N個訊息還沒有ack，則該consumer將block掉，直到有訊息ack。
global: true\false 是否將上面設定應用於channel，簡單點說，就是上面限制是channel級別還是consumer級別。
prefetchSize和global這兩項，rabbitmq沒有實現，暫且不研究prefetch_count在no_ask = false的情況下生效，即在自動應答的情況下這兩個值是不生效的。

第一個引數：訊息的限制大小，訊息多少兆。一般不做限制，設定為0
第二個引數：一次最多處理多少條，實際工作中設定為1就好
第三個引數：限流策略在什麼上應用。在RabbitMQ一般有兩個應用級別：1.通道 2.Consumer級別。一般設定為false，true 表示channel級別，false表示在consumer級別

2.2 程式碼演示

2.2.1 生產者


/**
 * 
* @ClassName: Producer 
* @Description: 生產者
* @author Coder程式設計
* @date2019年7月30日 下午23:15:51 
*
 */
public class Producer {

    
    public static void main(String[] args) throws Exception {
        
        //1 建立ConnectionFactory
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        
        String exchange = "test_qos_exchange";
        String routingKey = "qos.save";
        
        String msg = "Hello RabbitMQ QOS Message";
        
        for(int i =0; i<5; i ++){
            channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());
        }
        
    }
}

2.2.2 消費者


/**
 * 
* @ClassName: Consumer 
* @Description: 消費者
* @author Coder程式設計
* @date2019年7月30日 下午23:13:51 
*
 */
public class Consumer {

    
    public static void main(String[] args) throws Exception {
        
        
        //1 建立ConnectionFactory
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        
        
        String exchangeName = "test_qos_exchange";
        String queueName = "test_qos_queue";
        String routingKey = "qos.#";
        
        channel.exchangeDeclare(exchangeName, "topic", true, false, null);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);
        
        //1 限流方式  第一件事就是 autoAck設定為 false
        //設定為1，表示一條一條資料處理
        channel.basicQos(0, 1, false);
        
        channel.basicConsume(queueName, false, new MyConsumer(channel));
        
        
    }
}

2.2.3 自定義類：MyConsumer


/**
 * 
* @ClassName: MyConsumer 
* @Description: TODO
* @author Coder程式設計
* @date 2019年7月30日 下午23:11:55 
*
 */
public class MyConsumer extends DefaultConsumer {


    private Channel channel ;
    
    public MyConsumer(Channel channel) {
        super(channel);
        this.channel = channel;
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        System.err.println("-----------consume message----------");
        System.err.println("consumerTag: " + consumerTag);
        System.err.println("envelope: " + envelope);
        System.err.println("properties: " + properties);
        System.err.println("body: " + new String(body));
        
        //需要做簽收,false表示不支援批量簽收
        channel.basicAck(envelope.getDeliveryTag(), false);
        
    }


}

2.2.4 測試結果

我們先註釋掉：channel.basicAck(envelope.getDeliveryTag(), false);然後啟動Consumer。
檢視Exchange

檢視Queues

然後再啟動Producer。檢視列印結果：

我們會發現消費端，只收到了一條訊息。這是為什麼呢？

第一點因為我們在consumer中

channel.basicConsume(queueName, false, new MyConsumer(channel));

第二個引數設定為false為手動簽收。

第二點在qos中設定只接受一條訊息。如果這一條訊息不給Broker Ack應答的話，那麼Broker會認為你並沒有消費完這一條訊息，那麼就不會繼續傳送訊息。

channel.basicQos(0, 1, false);

可以看下管控臺，unack=1，Ready=4，total=5.

接下來我們放開註釋channel.basicAck(envelope.getDeliveryTag(), false); 進行訊息簽收。重啟服務。

3.1 列印結果

可以看到正常列印五條結果

4. 消費端ACK與重回佇列

4.1 消費端的手工ACK和NACK

消費端進行消費的時候，如果由於業務異常我們可以進行日誌的記錄，然後進行補償！

如果由於伺服器宕機等嚴重問題，那我們就需要手工進行ACK保障消費端消費成功！

4.2 消費端的重回佇列

消費端重回佇列是為了對沒有處理成功的訊息，把訊息重新傳遞給Broker!

一般我們在實際應用中，都會關閉重回佇列，也就是設定為False.

4.3 程式碼演示

4.3.1 生產者


/**
 * 
* @ClassName: Producer 
* @Description: 生產者
* @author Coder程式設計
* @date2019年7月30日 下午23:15:51 
*
 */
public class Producer {

    
    public static void main(String[] args) throws Exception {
        
        //1建立ConnectionFactory
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        
        String exchange = "test_ack_exchange";
        String routingKey = "ack.save";
        
        
        
        for(int i =0; i<5; i ++){
            
            Map<String, Object> headers = new HashMap<String, Object>();
            headers.put("num", i);
            
            //新增屬性，後續會使用到
            AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                    .deliveryMode(2) //投遞模式，持久化
                    .contentEncoding("UTF-8")
                    .headers(headers)
                    .build();
            String msg = "Hello RabbitMQ ACK Message " + i;
            channel.basicPublish(exchange, routingKey, true, properties, msg.getBytes());
        }
        
    }
}

4.3.2 消費者


/**
 * 
* @ClassName: Consumer 
* @Description: 消費者
* @author Coder程式設計
* @date2019年7月30日 下午23:13:51 
*
 */
public class Consumer {

    
    public static void main(String[] args) throws Exception {
        
        
        //1建立ConnectionFactory
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        
        
        String exchangeName = "test_ack_exchange";
        String queueName = "test_ack_queue";
        String routingKey = "ack.#";
        
        channel.exchangeDeclare(exchangeName, "topic", true, false, null);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);
        
        // 手工簽收 必須要關閉 autoAck = false
        channel.basicConsume(queueName, false, new MyConsumer(channel));
        
        
    }
}

4.3.3 自定義類：MyConsumer

/**
 * 
* @ClassName: MyConsumer 
* @Description: TODO
* @author Coder程式設計
* @date 2019年7月30日 下午23:11:55 
*
 */
public class MyConsumer extends DefaultConsumer {


    private Channel channel ;
    
    public MyConsumer(Channel channel) {
        super(channel);
        this.channel = channel;
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        System.err.println("-----------consume message----------");
        System.err.println("body: " + new String(body));
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if((Integer)properties.getHeaders().get("num") == 0) {
            //Nack三個引數  第二個引數：是否是批量，第三個引數：是否重回佇列（需要注意可能會發生重複消費，造成死迴圈）
            channel.basicNack(envelope.getDeliveryTag(), false, true);
        } else {
            channel.basicAck(envelope.getDeliveryTag(), false);
        }
        
    }


}

5.1 列印結果：

注意：
可以看到重回佇列會出現重複消費導致死迴圈的問題，這時候最好設定重試次數，比如超過三次後，訊息還是消費失敗，就將訊息丟棄。

6. TTL佇列/訊息

6.1 TTL

  • TTL是Time To Live的縮寫，也就是生存時間
  • RabbitMQ支援訊息的過期時間，在訊息傳送時可以進行指定
  • RabbitMQ支援佇列的過期時間，從訊息入佇列開始計算，只要超過了佇列的超時時間配置，那麼訊息會自動的清除

6.2 程式碼演示

6.2.1 直接通過管控臺進行演示

通過管控臺建立一個佇列

x-max-length 佇列的最大大小
x-message-ttl 設定10秒鐘，如果訊息還沒有被消費的話，就會被清除。

新增exchange

Queue與Exchange進行繫結

點選 test_ttl_exchange 進行繫結

檢視是否繫結成功

通過管控臺傳送訊息

訊息未處理自動清除

生產端設定過期時間

AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                .deliveryMode(2)
                .contentEncoding("UTF-8")
                .expiration("10000")
                .headers(headers)
                .build();

這兩個屬性並不相同，一個對應的是訊息體，一個對應的是佇列的過期。

7. 死信佇列

7.1 概念理解

死信佇列：DLX,Dead-Letter-Exchange
RabbitMQ的死信隊裡與Exchange息息相關

  • 利用DLX,當訊息在一個佇列中變成死信（dead message）之後，它能被重新publish到另一個Exchange，這個Exchange就是DLX

訊息變成死信有以下幾種情況

  • 訊息被拒絕（basic.reject/basic.nack）並且requeue=false
  • 訊息TTL過期
  • 佇列達到最大長度

DLX也是一個正常的Exchange，和一般的Exchange沒有區別，它能在任何的佇列上被指定，實際上就是設定某個佇列的屬性

當這個佇列中有死信時，RabbitMQ就會自動的將這個訊息重新發布到設定的Exchange上去，進而被路由到另一個佇列。

可以監聽這個佇列中訊息做相應的處理，這個特徵可以彌補RabbitMQ3.0以前支援的immediate引數的功能。

7.2 程式碼演示

  • 死信佇列設定：
  • 首先需要設定死信佇列的exchange和queue,然後進行繫結：
    Exchange:dlx.exchange
    Queue:dlx.queue
    RoutingKey:#
  • 然後我們進行正常宣告交換機、佇列、繫結，只不過我們需要在佇列加上一個引數即可：arguments.put("x-dead-letter-exchange","dlx.exchange");
  • 這樣訊息在過期、requeue、佇列在達到最大長度時，訊息就可以直接路由到死信佇列！

7.2.1 生產者

/**
 * 
* @ClassName: Producer 
* @Description: 生產者
* @author Coder程式設計
* @date2019年7月30日 下午23:15:51 
*
 */
public class Producer {

    
    public static void main(String[] args) throws Exception {
        
        //建立ConnectionFactory
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        
        String exchange = "test_dlx_exchange";
        String routingKey = "dlx.save";
        
        String msg = "Hello RabbitMQ DLX Message";
        
        for(int i =0; i<1; i ++){
            
            AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                    .deliveryMode(2)
                    .contentEncoding("UTF-8")
                    .expiration("10000")
                    .build();
            channel.basicPublish(exchange, routingKey, true, properties, msg.getBytes());
        }
        
    }
}

7.2.2 消費者

/**
 * 
* @ClassName: Consumer 
* @Description: 消費者
* @author Coder程式設計
* @date2019年7月30日 下午23:13:51 
*
 */
public class Consumer {

    
    public static void main(String[] args) throws Exception {
        
        
        //建立ConnectionFactory
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        
        // 這就是一個普通的交換機 和 佇列 以及路由
        String exchangeName = "test_dlx_exchange";
        String routingKey = "dlx.#";
        String queueName = "test_dlx_queue";
        
        channel.exchangeDeclare(exchangeName, "topic", true, false, null);
        
        Map<String, Object> agruments = new HashMap<String, Object>();
        agruments.put("x-dead-letter-exchange", "dlx.exchange");
        //這個agruments屬性，要設定到宣告佇列上
        channel.queueDeclare(queueName, true, false, false, agruments);
        channel.queueBind(queueName, exchangeName, routingKey);
        
        //要進行死信佇列的宣告:
        channel.exchangeDeclare("dlx.exchange", "topic", true, false, null);
        channel.queueDeclare("dlx.queue", true, false, false, null);
        channel.queueBind("dlx.queue", "dlx.exchange", "#");
        
        channel.basicConsume(queueName, true, new MyConsumer(channel));
        
        
    }
}

7.2.3 自定義類：MyConsumer


/**
 * 
* @ClassName: MyConsumer 
* @Description: TODO
* @author Coder程式設計
* @date 2019年7月30日 下午23:11:55 
*
 */
public class MyConsumer extends DefaultConsumer {


    public MyConsumer(Channel channel) {
        super(channel);
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        System.err.println("-----------consume message----------");
        System.err.println("consumerTag: " + consumerTag);
        System.err.println("envelope: " + envelope);
        System.err.println("properties: " + properties);
        System.err.println("body: " + new String(body));
    }
}

7.2.4 測試結果

執行Consumer，檢視管控臺
檢視Exchanges

檢視queue

可以看到test_dlx_queue多了DLX的標識，表示當佇列中出現死信的時候，會將訊息傳送到死信佇列dlx_queue中

關閉Consumer，只執行Producer

過10秒鐘後，訊息過期

在我們工作中，死信佇列非常重要，用於訊息沒有消費者，處於死信狀態。我們可以才用補償機制。

小結

本次主要介紹了RabbitMQ的高階特性，首先介紹了網際網路大廠在實際使用中如何保障100%的訊息投遞成功和冪等性的，以及對RabbitMQ的確認訊息、返回訊息、ACK與重回佇列、訊息的限流，以及對超時時間、死信佇列的使用

文末

歡迎關注個人微信公眾號：Coder程式設計
獲取最新原創技術文章和免費學習資料，更有大量精品思維導圖、面試資料、PMP備考資料等你來領，方便你隨時隨地學習技術知識！
新建了一個qq群：315211365，歡迎大家進群交流一起學習。謝謝了！也可以介紹給身邊有需要的朋友。

文章收錄至
Github: https://github.com/CoderMerlin/coder-programming
Gitee: https://gitee.com/573059382/coder-programming
歡迎關注並star~

參考文章：

《RabbitMQ訊息中介軟體精講》

推薦文章：

訊息中介軟體——RabbitMQ（五）快速入門生產者與消費者，SpringBoot整合RabbitMQ！

訊息中介軟體——RabbitMQ（六）理解Exchange交換機核心概念!

訊息中介軟體——RabbitMQ（七）高階特性全在這裡!（上

相關推薦

訊息中介軟體——RabbitMQ（八）高階特性全在這裡!（下）

前言 上一篇訊息中介軟體——RabbitMQ（七）高階特性全在這裡!（上）中我們介紹了訊息如何保障100%的投遞成功？,冪等性概念詳解,在海量訂單產生的業務高峰期，如何避免訊息的重複消費的問題？,Confirm確認訊息、Return返回訊息。這篇我們來介紹下下面內容。 自定義消費者 訊息的限流（防止

訊息中介軟體--RabbitMQ學習（十七）---高階特性之死信佇列

死信佇列:DLX,Dead- Letter- Exchange 利用DLX,當訊息在一個佇列中變成死信( dead message)之後它能被重新 publish到另一個 Exchange,這個 Exchange就是DLX 死信佇列訊息變成死信有一下幾種情況

訊息中介軟體——RabbitMQ（七）高階特性全在這裡!（上）

前言 前面我們介紹了RabbitMQ的安裝、各大訊息中介軟體的對比、AMQP核心概念、管控臺的使用、快速入門RabbitMQ。本章將介紹RabbitMQ的高階特性。分兩篇（上/下）進行介紹。 訊息如何保障100%的投遞成功？ 冪等性概念詳解 在海量訂單產生的業務高峰期，如何避免訊息的重複消費的問題？

訊息中介軟體--RabbitMQ學習（一）

Activemq介紹 Activemq是 Apache出品,最流行的能力強勁的開源訊息匯流排,並且它個完全支援MS規範的訊息中介軟體。 其豐富的AP、多種叢集構建模式使得他成為業界老牌訊息中介軟體,在中小型企業中應用廣泛。 MQ衡量指標:服務效能、資料儲存、叢集架構

訊息中介軟體--RabbitMQ學習（二）

Server:又稱 Broker,接受客戶端的連線,實現AMQP實體服務。 Connection:連線,應用程式與 Broker的網路連線。 Channel:網路通道,幾乎所有的操作都在 Channel中進行, Channel是進行訊息讀寫的通道。客戶端可建立多個 hannel,每個 Channel代表一個

訊息中介軟體--RabbitMQ學習（六）

Fanout Exchange學習 Fanout Exchange介紹 不處理路由鍵,只需要簡單的將佇列繫結到交換機上 傳送到交換機的訊息都會被轉發到與該交換機繫結的所有佇列上 Fanout交換機轉發訊息是最快的 只要交換機跟佇列有繫結，就能夠傳送訊息過去。

十次方專案第五天（訊息中介軟體RabbitMQ）

學習目標： 能夠說出訊息佇列的應用場景以及RabbitMQ的主要概念 完成RabbitMQ安裝以及RabbitMQ三種模式的入門案例 完成使用者註冊，能夠將訊息傳送給RabbitMQ 完成簡訊微服務，能夠接收訊息並呼叫阿里雲通訊完成簡訊傳送 1 RabbitMQ簡介

訊息中介軟體Rabbitmq（二）-使用詳解

https://blog.csdn.net/Dante_003/article/details/79377908Rabbitmq 是基於amqp（高階訊息佇列協議）實現的佇列技術，在他之上可以完成多種型別的訊息轉發模型。 下面列舉一些常用的訊息轉發場景，在rabbitmq中是

訊息中介軟體——RabbitMQ（一）Windows/Linux環境搭建（完整版）

前言 最近在學習訊息中介軟體——RabbitMQ，打算把這個學習過程記錄下來。此章主要介紹環境搭建。此次主要是單機搭建（條件有限），包括在Windows、Linux環境下的搭建，以及RabbitMQ的監控平臺搭建。 環境準備 在搭建RabbitMQ之前，請先確保如下環境已經搭建完畢 Java環境（我

訊息中介軟體——RabbitMQ（二）各大主流訊息中介軟體綜合對比介紹！

前言 訊息佇列已經逐漸成為企業IT系統內部通訊的核心手段。它具有低耦合、可靠投遞、廣播、流量控制、最終一致性等一系列功能，成為非同步RPC的主要手段之一。當今市面上有很多主流的訊息中介軟體，如老牌的ActiveMQ、RabbitMQ，炙手可熱的Kafka，阿里巴巴自主開發RocketMQ等。今天主要來

訊息中介軟體——RabbitMQ（三）理解RabbitMQ核心概念和AMQP協議！

前言 本章學習，我們可以瞭解到以下知識點： 網際網路大廠為什麼選擇RabbitMQ？ RabbiMQ的高效能之道是如何做到的？ 什麼是AMQP高階協議？ AMQP核心概念是什麼？ RabbitMQ整體架構模型是什麼樣子的？ RabbitMQ訊息是如何流轉的？ 1. 初識RabbitMQ Rabbi

訊息中介軟體——RabbitMQ（四）命令列與管控臺的基本操作！

前言 在前面的文章中我們介紹過RabbitMQ的搭建：RabbitMQ的安裝過以及各大主流訊息中介軟體的對比：，本章就主要來介紹下我們之前安裝的管控臺是如何使用以及如何通過命令列進行操作。 1. 命令列操作 1.1 基礎服務的命令操作 rabbitmqctl stop_app：關閉應用 rabbitm

訊息中介軟體——RabbitMQ（五）快速入門生產者與消費者，SpringBoot整合RabbitMQ！

前言 本章我們來一次快速入門RabbitMQ——生產者與消費者。需要構建一個生產端與消費端的模型。什麼意思呢？我們的生產者傳送一條訊息，投遞到RabbitMQ叢集也就是Broker。 我們的消費端進行監聽RabbitMQ，當發現佇列中有訊息後，就進行消費。 1. 環境準備 本次整合主要採用Spring

訊息中介軟體——RabbitMQ（六）理解Exchange交換機核心概念！

前言 來了解RabbitMQ一個重要的概念：Exchange交換機 1. Exchange概念 Exchange：接收訊息，並根據路由鍵轉發訊息所繫結的佇列。 藍色框：客戶端傳送訊息至交換機，通過路由鍵路由至指定的佇列。 黃色框：交換機和佇列通過路由鍵有一個繫結的關係。 綠色框：消費端通過監聽

訊息中介軟體—RabbitMQ（叢集原理與搭建篇)

一般來說，如果只是為了學習RabbitMQ或者驗證業務工程的正確性那麼在本地環境或者測試環境上使

JMS訊息中介軟體原理及ActiveMQ在企業中的應用（接上篇）

程式碼實現：傳送訊息---》接受訊息---》伺服器配置 //1 傳送訊息（接受回覆訊息） public class SenderMessageService { //釋出指定訊息到指定地址(在釋出之前，建議將訊息儲存到資料庫)  public void publish(St

訊息中介軟體——RabbitMQ的高階特性

前言 前面我們介紹了RabbitMQ的安裝、各大訊息中介軟體的對比、AMQP核心概念、管控臺的使用、快速入門RabbitMQ。

訊息中介軟體rabbitmq(1)

介紹 注：沒有註冊中心，總共只分為三個角色，product,server,customer。 Queue 訊息佇列，等同於rocketmq中的topic Exchange 生產者產生的訊息並不是直接傳送給訊息佇列Queue的，而是要經過Exchange（交換器），由Exchange再將訊息路由到

訊息中介軟體rabbitmq(2)

介紹 主要演示生產者和消費如何傳遞訊息的 生產者 public static void main(String[] args) throws IOException, TimeoutException { // TODO Auto-generated method stub

訊息中介軟體RabbitMQ初使用

一、Docker下安裝RabbitMQ 1、進docker網站，查詢RabbitMQ映象。獲取字尾帶management的映象，因為此種映象帶web管理頁面。 https://hub.docker.com/ 2、使用docker命令獲取 docker pull regi