一、RabbitMQ簡介

1、什麼是RabbitMQ

AMQP，即Advanced Message Queuing Protocol，高階訊息佇列協議，是應用層協議的一個開放標準，為面向訊息的中介軟體設計。訊息中介軟體主要用於元件之間的解耦，訊息的傳送者無需知道訊息使用者的存在，反之亦然。AMQP的主要特徵是面向訊息、佇列、路由（包括點對點和釋出/訂閱）、可靠性、安全。
RabbitMQ是一個開源的AMQP實現，伺服器端用Erlang語言編寫，支援多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支援AJAX。用於在分散式系統中儲存轉發訊息，在易用性、擴充套件性、高可用性等方面表現不俗。

2、RabbitMQ基礎概念

• ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel都是RabbitMQ對外提供的API中最基本的物件。Connection是RabbitMQ的socket連結，它封裝了socket協議相關部分邏輯。ConnectionFactory為Connection的製造工廠。
Channel是我們與RabbitMQ打交道的最重要的一個介面，我們大部分的業務操作是在Channel這個介面中完成的，包括定義Queue、定義Exchange、繫結Queue與Exchange、釋出訊息等。

• Queue

Queue（佇列）是RabbitMQ的內部物件，用於儲存訊息，用下圖表示。

RabbitMQ中的訊息都只能儲存在Queue中，生產者（下圖中的P）生產訊息並最終投遞到Queue中，消費者（下圖中的C）可以從Queue中獲取訊息並消費。

多個消費者可以訂閱同一個Queue，這時Queue中的訊息會被平均分攤給多個消費者進行處理，而不是每個消費者都收到所有的訊息並處理。

• Message acknowledgment

在實際應用中，可能會發生消費者收到Queue中的訊息，但沒有處理完成就宕機（或出現其他意外）的情況，這種情況下就可能會導致訊息丟失。為了避免這種情況發生，我們可以要求消費者在消費完訊息後傳送一個回執給RabbitMQ，RabbitMQ收到訊息回執（Message acknowledgment）後才將該訊息從Queue中移除；
如果RabbitMQ沒有收到回執並檢測到消費者的RabbitMQ連線斷開，則RabbitMQ會將該訊息傳送給其他消費者（如果存在多個消費者）進行處理。這裡不存在timeout概念，一個消費者處理訊息時間再長也不會導致該訊息被髮送給其他消費者，除非它的RabbitMQ連線斷開。
這裡會產生另外一個問題，如果我們的開發人員在處理完業務邏輯後，忘記傳送回執給RabbitMQ，這將會導致嚴重的bug——Queue中堆積的訊息會越來越多；消費者重啟後會重複消費這些訊息並重復執行業務邏輯。另外pub message是沒有ack的。

• Message durability

如果我們希望即使在RabbitMQ服務重啟的情況下，也不會丟失訊息，我們可以將Queue與Message都設定為可持久化的（durable），這樣可以保證絕大部分情況下我們的RabbitMQ訊息不會丟失。但依然解決不了小概率丟失事件的發生（比如RabbitMQ伺服器已經接收到生產者的訊息，但還沒來得及持久化該訊息時RabbitMQ伺服器就斷電了），如果我們需要對這種小概率事件也要管理起來，那麼我們要用到事務。由於這裡僅為RabbitMQ的簡單介紹，所以這裡將不講解RabbitMQ相關的事務。

• Prefetch count

前面我們講到如果有多個消費者同時訂閱同一個Queue中的訊息，Queue中的訊息會被平攤給多個消費者。這時如果每個訊息的處理時間不同，就有可能會導致某些消費者一直在忙，而另外一些消費者很快就處理完手頭工作並一直空閒的情況。我們可以通過設定prefetchCount來限制Queue每次傳送給每個消費者的訊息數，比如我們設定prefetchCount=1，則Queue每次給每個消費者傳送一條訊息；消費者處理完這條訊息後Queue會再給該消費者傳送一條訊息。

• Exchange

在上一節我們看到生產者將訊息投遞到Queue中，實際上這在RabbitMQ中這種事情永遠都不會發生。實際的情況是，生產者將訊息傳送到Exchange（交換器，下圖中的X），由Exchange將訊息路由到一個或多個Queue中（如果沒有Queue與之繫結，則訊息會被丟棄）。

Exchange是按照什麼邏輯將訊息路由到Queue的？這個將在Binding一節介紹。
RabbitMQ中的Exchange有四種類型，不同的型別有著不同的路由策略，這將在Exchange Types一節介紹。

• routing key

生產者在將訊息傳送給Exchange的時候，一般會指定一個routing key，來指定這個訊息的路由規則，而這個routing key需要與Exchange Type及binding key聯合使用才能最終生效。
在Exchange Type與binding key固定的情況下（在正常使用時一般這些內容都是固定配置好的），我們的生產者就可以在傳送訊息給Exchange時，通過指定routing key來決定訊息流向哪裡。
RabbitMQ為routing key設定的長度限制為255 bytes。

• Binding

RabbitMQ中通過Binding將Exchange與Queue關聯起來，這樣RabbitMQ就知道如何正確地將訊息路由到指定的Queue了。

• Binding key

在繫結（Binding）Exchange與Queue的同時，一般會指定一個binding key；消費者將訊息傳送給Exchange時，一般會指定一個routing key；當binding key與routing key相匹配時，訊息將會被路由到對應的Queue中。這個將在Exchange Types章節會列舉實際的例子加以說明。
在繫結多個Queue到同一個Exchange的時候，這些Binding允許使用相同的binding key。
binding key 並不是在所有情況下都生效，它依賴於Exchange Type，比如fanout型別的Exchange就會無視binding key，而是將訊息路由到所有繫結到該Exchange的Queue。

• Exchange Types

RabbitMQ常用的Exchange Type有fanout、direct、topic、headers這四種（AMQP規範裡還提到兩種Exchange Type，分別為system與自定義，這裡不予以描述），下面分別進行介紹。
1、fanout
fanout型別的Exchange路由規則非常簡單，它會把所有傳送到該Exchange的訊息路由到所有與它繫結的Queue中。

上圖中，生產者（P）傳送到Exchange（X）的所有訊息都會路由到圖中的兩個Queue，並最終被兩個消費者（C1與C2）消費。

2、direct
direct型別的Exchange路由規則也很簡單，它會把訊息路由到那些binding key與routing key完全匹配的Queue中。

以上圖的配置為例，我們以routingKey=”error”傳送訊息到Exchange，則訊息會路由到Queue1（amqp.gen-S9b…，這是由RabbitMQ自動生成的Queue名稱）和Queue2（amqp.gen-Agl…）；如果我們以routingKey=”info”或routingKey=”warning”來發送訊息，則訊息只會路由到Queue2。如果我們以其他routingKey傳送訊息，則訊息不會路由到這兩個Queue中。

3、topic
前面講到direct型別的Exchange路由規則是完全匹配binding key與routing key，但這種嚴格的匹配方式在很多情況下不能滿足實際業務需求。topic型別的Exchange在匹配規則上進行了擴充套件，它與direct型別的Exchage相似，也是將訊息路由到binding key與routing key相匹配的Queue中，但這裡的匹配規則有些不同，它約定：
routing key為一個句點號“. ”分隔的字串（我們將被句點號“. ”分隔開的每一段獨立的字串稱為一個單詞），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”
binding key與routing key一樣也是句點號“. ”分隔的字串。
binding key中可以存在兩種特殊字元“”與“#”，用於做模糊匹配。其中“”用於匹配一個單詞，“#”用於匹配多個單詞（可以是零個）

以上圖中的配置為例，routingKey=”quick.orange.rabbit”的訊息會同時路由到Q1與Q2，routingKey=”lazy.orange.fox”的訊息會路由到Q1與Q2，routingKey=”lazy.brown.fox”的訊息會路由到Q2，routingKey=”lazy.pink.rabbit”的訊息會路由到Q2（只會投遞給Q2一次，雖然這個routingKey與Q2的兩個bindingKey都匹配）；routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的訊息將會被丟棄，因為它們沒有匹配任何bindingKey。

4、headers
headers型別的Exchange不依賴於routing key與binding key的匹配規則來路由訊息，而是根據傳送的訊息內容中的headers屬性進行匹配。
在繫結Queue與Exchange時指定一組鍵值對；當訊息傳送到Exchange時，RabbitMQ會取到該訊息的headers（也是一個鍵值對的形式），對比其中的鍵值對是否完全匹配Queue與Exchange繫結時指定的鍵值對；如果完全匹配則訊息會路由到該Queue，否則不會路由到該Queue。
該型別的Exchange沒有用到過（不過也應該很有用武之地），所以不做介紹。

• RPC

MQ本身是基於非同步的訊息處理，前面的示例中所有的生產者（P）將訊息傳送到RabbitMQ後不會知道消費者（C）處理成功或者失敗（甚至連有沒有消費者來處理這條訊息都不知道）。
但實際的應用場景中，我們很可能需要一些同步處理，需要同步等待服務端將我的訊息處理完成後再進行下一步處理。這相當於RPC（Remote Procedure Call，遠端過程呼叫）。在RabbitMQ中也支援RPC。

RabbitMQ中實現RPC的機制是：
客戶端傳送請求（訊息）時，在訊息的屬性（MessageProperties，在AMQP協議中定義了14中properties，這些屬性會隨著訊息一起傳送）中設定兩個值replyTo（一個Queue名稱，用於告訴伺服器處理完成後將通知我的訊息傳送到這個Queue中）和correlationId（此次請求的標識號，伺服器處理完成後需要將此屬性返還，客戶端將根據這個id瞭解哪條請求被成功執行了或執行失敗）。伺服器端收到訊息並處理，伺服器端處理完訊息後，將生成一條應答訊息到replyTo指定的Queue，同時帶上correlationId屬性。客戶端之前已訂閱replyTo指定的Queue，從中收到伺服器的應答訊息後，根據其中的correlationId屬性分析哪條請求被執行了，根據執行結果進行後續業務處理。

二、RabbitMQ的客戶端amqp-client

通過amqp-client來操作RabbitMQ，在pom檔案引入如下依賴即可。

        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>3.4.1</version>
        </dependency>

然後定義一個RabbitMQ的工具類，用於建立連線。

package com.kang.rabbitmq.util;

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

public class ConnectionUtil {

    public static Connection getConnection() throws Exception {
        //定義連線工廠
        ConnectionFactory factory = new ConnectionFactory();
        //設定服務地址
        factory.setHost("127.0.0.1");
        //埠
        factory.setPort(5672);
        //設定賬號資訊，使用者名稱、密碼、vhost
        factory.setVirtualHost("/enjoyshop");
        factory.setUsername("enjoyshop");
        factory.setPassword("enjoyshop");
        // 通過工程獲取連線
        Connection connection = factory.newConnection();
        return connection;
    }

}

相關API：
1、連線

ConnectionFactory factory = new ConnectionFactory();
factory.setHost(hostName);
Connection conn = factory.newConnection();
Channel channel = conn.createChannel();

以上是得到一個rabbitmq連線最最基礎的程式碼，當然了，還可以設定一些諸如使用者名稱密碼的事情。最後這個channel就可以用來收和發訊息了。
2、訊息者執行緒池

ExecutorService es = Executors.newFixedThreadPool(20);
Connection conn = factory.newConnection(es);

消費者時使用，上述自動開了一20個執行緒的池來搞。

3、地址陣列

Address[] addrArr = new Address[]{ 
new Address(hostname1, portnumber1), 
new Address(hostname2, portnumber2)};
Connection conn = factory.newConnection(addrArr);

上述程式碼如果連hostname1失敗了就去hostname2。
factory.newConnection()會觸發這個檢測。

4、宣告exchange與queue

channel.exchangeDeclare(exchangeName, "direct", true);
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, exchangeName, routingKey);

channel.exchangeDeclare 方法引數有：交換機名字（exchangeName） 、型別（“direct”）、 是否持久化 （true）、不使用時是否自動刪除、是否是內部的（不能被客戶端使用）以及其他引數。
channel.queueDeclare 方法引數有:：queue名字、是否持久化、是否是獨佔的queue（僅供此連線）、不使用時是否自動刪除和其他引數。
channel.queueBind 引數有 ：queue名字、交換機名字、此次繫結使用的路由關鍵字和其他引數。

5、發出訊息

byte[] messageBodyBytes = "Hello, world!".getBytes();
channel.basicPublish(exchangeName, routingKey, null, messageBodyBytes);

channel.basicPublish 引數有:要發出的交換機名字、路由關鍵字、是否強制（設定為true時，找不到收的人時可以通過returnListener返回）、是否立即（其實rabbitmq不支援）、其他屬性 ，最後是訊息主體。

6、執行緒安全
Channel是執行緒安全的，但是最好是每個執行緒裡用自己的Channel，因為在單個Channel裡排隊是有可能慢一些的。

三、RabbitMQ的工作模式示例

1、簡單佇列模式

P：訊息的生產者
C：訊息的消費者
紅色：佇列
生產者將訊息傳送到佇列，消費者從佇列中獲取訊息。

• 生產者傳送訊息到佇列：

package com.kang.rabbitmq.simple;

import com.kang.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
//簡單模式
public class Send {

    private final static String QUEUE_NAME = "test_queue";

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

        // 訊息內容
        String message = "Hello World!";
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");

        //關閉通道和連線
        channel.close();
        connection.close();
    }
}

• 消費者從佇列中獲取訊息：

package com.kang.rabbitmq.simple;

import com.kang.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
//簡單模式
public class Recv {

    private final static String QUEUE_NAME = "test_queue";

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

        // 定義佇列的消費者
        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(" [x] Received '" + message + "'");
        }
    }
}

2、Work模式

這裡有一個生產者、2個消費者。
一個訊息只能被一個消費者獲取。

• 消費者1

package com.kang.rabbitmq.work;

import com.kang.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
//work模式
public class Recv {

    private final static String QUEUE_NAME = "test_queue_work";

    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.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(" [x] Received '" + message + "'");
            // 休眠
            Thread.sleep(10);
            // 返回確認狀態
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }
}

• 消費者2

package com.kang.rabbitmq.work;

import com.kang.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
//work模式
public class Recv2 {

    private final static String QUEUE_NAME = "test_queue_work";

    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.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(" [x] Received '" + message + "'");
            // 休眠1秒
            Thread.sleep(1000);

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

• 生產者

package com.kang.rabbitmq.work;

import com.kang.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
//work模式
public class Send {

    private final static String QUEUE_NAME = "test_queue_work";

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

        for (int i = 0; i < 50; i++) {
            // 訊息內容
            String message = "" + i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");

            Thread.sleep(i * 10);
        }

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

• 程式執行結果：

1、消費者1和消費者2獲取到的訊息內容是不同的，同一個訊息只能被一個消費者獲取。
2、消費者1和消費者2獲取到的訊息的數量是相同的，一個是奇數一個是偶數。
這樣是不合理的，因為消費者1的休眠間隔更短，應該是消費者1要比消費者2獲取到的訊息多合理。

• Work模式的“能者多勞”

如果要實現上述功能，可以在每個消費者中加入如下程式碼：

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

        // 加入這一行程式碼使得同一時刻伺服器只會發一條訊息給消費者（即能者多勞）
        channel.basicQos(1);

        // 定義佇列的消費者
        QueueingConsumer consumer = new QueueingConsumer(channel);

這樣再執行程式碼的話，消費者1接受的訊息數就比消費者2的數量要多。

• 訊息的確認模式

消費者從佇列中獲取訊息，服務端如何知道訊息已經被消費呢？

模式1：自動確認
只要訊息從佇列中獲取，無論消費者獲取到訊息後是否成功訊息，都認為是訊息已經成功消費。可以通過如下程式碼設定：

// 監聽佇列
channel.basicConsume(QUEUE_NAME, true, consumer);

basicConsume方法的第二個引數設為true，即表示採用自動確認模式。

模式2：手動確認
消費者從佇列中獲取訊息後，伺服器會將該訊息標記為不可用狀態，等待消費者的反饋，如果消費者一直沒有反饋，那麼該訊息將一直處於不可用狀態。可以通過如下程式碼設定：

// 監聽佇列，手動返回完成
channel.basicConsume(QUEUE_NAME, false, consumer);

確認程式碼如下：

// 返回確認狀態        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

3、訂閱模式

• 特點：

1、1個生產者，多個消費者。
2、每一個消費者都有自己的一個佇列。
3、生產者沒有將訊息直接傳送到佇列，而是傳送到了交換機。
4、每個佇列都要繫結到交換機。
5、生產者傳送的訊息，經過交換機，到達佇列，實現一個訊息被多個消費者獲取的目的。

• 訊息的生產者

package com.kang.rabbitmq.ps;

import com.kang.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
//訂閱模式
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 = "Hello World!";
        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");

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

注意：訊息傳送到沒有佇列繫結的交換機時，訊息將丟失，因為，交換機沒有儲存訊息的能力，訊息只能存在在佇列中。

• 訊息的消費者1

package com.kang.rabbitmq.ps;

import com.kang.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
//訂閱模式
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, false, consumer);

        // 獲取訊息
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" [x] Received '" + message + "'");
            Thread.sleep(10);

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

• 訊息的消費者2

package com.kang.rabbitmq.ps;

import com.kang.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
//訂閱模式
public class Recv2 {

    private final static String QUEUE_NAME = "test_queue_fanout_2";

    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, false, consumer);

        // 獲取訊息
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" [x] Received '" + message + "'");
            Thread.sleep(10);

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

4、路由模式

• 生產者

package com.kang.rabbitmq.routing;

import com.kang.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
//路由模式
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 = "Hello World!";
        channel.basicPublish(EXCHANGE_NAME, "key2", null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");

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

• 消費者1

package com.kang.rabbitmq.routing;

import com.kang.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
//路由模式
public class Recv {

    private final static String QUEUE_NAME = "test_queue_work";

    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, "key");
        //當需要繫結多個key值時，可以多次執行繫結語句
        //channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key3");

        // 同一時刻伺服器只會發一條訊息給消費者
        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(" [x] Received '" + message + "'");
            Thread.sleep(10);

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

• 消費者2

package com.kang.rabbitmq.routing;

import com.kang.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
//路由模式
public class Recv2 {

    private final static String QUEUE_NAME = "test_queue_work2";

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

        // 同一時刻伺服器只會發一條訊息給消費者
        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(" [x] Received '" + message + "'");
            Thread.sleep(10);

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

• 測試結果：

只有消費者2接收到了訊息，因為消費者2和生產者有相應的關鍵字繫結。

5、萬用字元模式

• 生產者

package com.kang.rabbitmq.topic;

import com.kang.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
//萬用字元模式
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 = "Hello World!";
        channel.basicPublish(EXCHANGE_NAME, "key.1", null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");

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

• 消費者1

package com.kang.rabbitmq.topic;

import com.kang.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
//萬用字元模式
public class Recv {

    private final static String QUEUE_NAME = "test_queue_topic_work";

    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, "key.*");

        // 同一時刻伺服器只會發一條訊息給消費者
        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(" [x] Received '" + message + "'");
            Thread.sleep(10);

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

• 消費者2

package com.kang.rabbitmq.topic;

import com.kang.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.QueueingConsumer;
//萬用字元模式
public class Recv2 {

    private final static String QUEUE_NAME = "test_queue_topic_work2";

    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, "*.*");

        // 同一時刻伺服器只會發一條訊息給消費者
        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(" [x] Received '" + message + "'");
            Thread.sleep(10);

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

• 測試結果：

兩個消費者都會收到訊息，因為”key.1”可以被兩者匹配。

四、Spring整合RabbitMQ

1、引入依賴

        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
            <version>1.4.0.RELEASE</version>
        </dependency>

2、編寫一個處理訊息的類

package com.kang.rabbitmq.spring;


public class Foo {

    //具體執行業務的方法
    public void listen(String foo) {
        System.out.println("消費者： " + foo);
    }
}

3、新增Spring整合配置檔案

<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    xsi:schemaLocation="http://www.springframework.org/schema/rabbit
    http://www.springframework.org/schema/rabbit/spring-rabbit-1.4.xsd
    http://www.springframework.org/schema/beans
    http://www.springframework.org/schema/beans/spring-beans-4.0.xsd">

    <!-- 定義RabbitMQ的連線工廠 -->
    <rabbit:connection-factory id="connectionFactory"
        host="127.0.0.1" port="5672" username="enjoyshop" password