譯:3.RabbitMQ Java Client 之 Publish/Subscribe（發佈和訂閱）

在上篇 RabbitMQ 之Work Queues (工做隊列) 教程中，咱們建立了一個工做隊列，工做隊列背後的假設是每一個任務都交付給一個工做者。

在這一部分，咱們將作一些徹底不一樣的事情 - 咱們將向多個消費者傳遞信息。此模式稱爲「發佈/訂閱」。

這篇爲譯文加上本身的理解，英文原文請移步：http://www.rabbitmq.com/tutorials/tutorial-three-java.html

---

爲了說明這種模式，咱們將構建一個簡單的日誌記錄系統。

它將包含兩個程序 - 第一個將發出日誌消息，第二個將接收和打印它們。

在咱們的日誌記錄系統中，接收程序的每一個運行副本都將獲取消息。

這樣咱們就能夠運行一個接收器並將日誌定向到磁盤; 同時咱們將可以運行另外一個接收器並在屏幕上看到日誌。

基本上，發佈的日誌消息將被廣播給全部接收者。

Exchanges 交換

在本教程的前幾部分中，咱們向隊列發送消息和從隊列接收消息。如今是時候在Rabbit中引入完整的消息傳遞模型了。

讓咱們快速回顧一下前面教程中介紹的內容：

• 生產者是發送消息的用戶的應用程序。
• 隊列是存儲消息的緩衝器。
• 消費者是接收消息的用戶的應用程序。

RabbitMQ中消息傳遞模型的核心思想是生產者永遠不會將任何消息直接發送到隊列。實際上，生產者一般甚至不知道消息是否會被傳遞到任何隊列。

相反，生產者只能向交換器發送消息。交換是一件很是簡單的事情。一方面，它接收來自生產者的消息，另外一方面將它們推送到隊列。

交換所必須確切知道如何處理收到的消息。它應該附加到特定隊列嗎？它應該附加到許多隊列嗎？或者它應該被丟棄。其規則由交換類型定義 。

 

Tips: 能夠看出，這節課咱們多了一個Exchanges ，生產者產生的消息將再也不直接發送給隊列，而是由Exchange來處理這件事情。

有幾種交換類型可供選擇：direct, topic, headers and fanout. 咱們將專一於最後這個-- fanout.  

讓咱們建立一個這種類型的交換，並將其稱爲日誌：

channel.exchangeDeclare（「logs」，「fanout」）;

 

 fanout (扇出交換)很是簡單。 正如您可能從名稱中猜到的那樣，它只是將收到的全部消息廣播到它知道的全部隊列中。而這正是咱們記錄器所須要的。

Listing exchanges

要列出服務器上的交換，您能夠運行有用的rabbitmqctl：

Linux 執行下列命令

sudo rabbitmqctl list_exchanges

 

Windows 執行下列命令

rabbitmqctl list_exchanges

在這個列表中有一些 amq.*  exchanges（交換） 和一些默認的 (沒有命名的) exchange（交換）

他們是默認建立的，可是你可能不須要使用他們如今。

Nameless exchange 無名交換

在本教程的前幾部分中，咱們對交換一無所知，但仍可以向隊列發送消息。 這是可能的，由於咱們使用的是默認交換，咱們經過空字符串（「」）來識別。

回想一下咱們以前是如何發佈消息的：

channel.basicPublish("", "hello", null, message.getBytes());

第一個參數是交換的名稱。 空字符串表示默認或無名交換：消息被路由到具備routingKey指定名稱的隊列（若是存在）

如今，咱們能夠發佈到咱們的命名交換：

channel.basicPublish( "logs", "", null, message.getBytes());

Temporary queues 臨時隊列

您可能還記得之前咱們使用的是具備指定名稱的隊列（請記住hello和task_queue？）。

可以命名隊列對咱們來講相當重要 - 咱們須要將工做人員指向同一個隊列。當您想要在生產者和消費者之間共享隊列時，爲隊列命名很重要。

但咱們的記錄器並不是如此。咱們但願瞭解全部日誌消息，而不只僅是它們的一部分。咱們也只對目前流動的消息感興趣，而不是舊消息。要解決這個問題，咱們須要兩件事。

首先，每當咱們鏈接到Rabbit時，咱們都須要一個新的空隊列。爲此，咱們可使用隨機名稱建立隊列，或者更好 - 讓服務器爲咱們選擇隨機隊列名稱。

其次，一旦咱們斷開消費者，就應該自動刪除隊列。

在Java客戶端中，當咱們沒有向queueDeclare（）提供參數時，咱們 使用生成的名稱建立一個非持久的，獨佔的自動刪除隊列：

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

 你也能夠學習更多關於 exclusive flag和其餘隊列屬性 在 guide on queues.

此時，queueName包含一個隨機隊列名稱。例如，它可能看起來像amq.gen-JzTY20BRgKO-HjmUJj0wLg。

Bindings 綁定

 

咱們已經建立了一個扇出交換和一個隊列。

如今咱們須要告訴交換機將消息發送到咱們的隊列。交換和隊列之間的關係稱爲綁定。

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

從如今開始，日誌交換會將消息附加到咱們的隊列中。

Listing bindings 列出綁定列表

rabbitmqctl list_bindings

Putting it all together 總體看下

生成日誌消息的生產者程序與前一個教程沒有太大的不一樣。

最重要的變化是咱們如今想要將消息發佈到咱們的日誌交換而不是無名交換。

Tips:

這裏簡單談下個人理解：

假設P是咱們平時工做的領導，X是祕書（某任務自動分配系統），C1 是員工張三，C2 是員工李四，

領導制定（發佈）好任務列表後，交給祕書（X, 任務分配系統（Exchange）），祕書（X, 任務分配 系統Exchange）將任務發送到這兩個郵箱（消息隊列）中便可。

張三，李四都綁定（訂閱）了不一樣的郵箱（不一樣的隊列名稱），那麼張三和李四取消息便從本身綁定的郵箱（隊列）中取便可。

上篇博文中的工做隊列所謂的無名交換能夠理解爲沒有祕書（exchange）這個角色，並且共用同一個消息隊列，如此而已。

 

咱們須要在發送時提供routingKey，可是對於扇出交換，它的值會被忽略。這裏是EmitLog.java程序的代碼 ：

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

public class EmitLog {
    private static final String EXCHANGE_NAME = "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, BuiltinExchangeType.FANOUT);
        String message = getMessage(argv);
        channel.basicPublish(EXCHANGE_NAME,"", null, message.getBytes("UTF-8"));
        System.out.println(" [x] Sent '" + message + "'");
        channel.close();
        connection.close();
    }

    private static String getMessage(String[] strings) {
        if (strings.length < 1)
            return "info: Hello World!";
        return joinStrings(strings, " ");
    }

    private static String joinStrings(String[] strings, String delimiter) {
        int length = strings.length;
        if (length == 0)
            return "";
        StringBuilder words = new StringBuilder(strings[0]);
        for (int i = 1; i < length; i++) {
            words.append(delimiter).append(strings[i]);
        }
        return words.toString();
    }
}

 

如您所見，在創建鏈接後咱們宣佈了交換。此步驟是必要的，由於禁止發佈到不存在的交換。

若是沒有隊列綁定到交換機，消息將會丟失，但這對咱們沒有問題; 若是沒有消費者在聽，咱們能夠安全地丟棄該消息。

 ReceiveLogs.java的代碼：

import java.io.IOException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

public class ReceiveLogs {

    private static final String EXCHANGE_NAME = "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, BuiltinExchangeType.FANOUT);
        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName, EXCHANGE_NAME, "");
        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 '" + message + "'");
            }
        };
        channel.basicConsume(queueName, true, consumer);
    }
}

 使用rabbitmqctl list_bindings，您能夠驗證代碼是否實際建立了咱們想要的綁定和隊列。 

rabbitmqctl list_bindings

 運行兩個ReceiveLogs.java程序時，您應該看到以下內容：

 

Tips: amq.gen-JzTY20BRgKO-HjmUJj0wLg 是隨機生成的隊列名稱。 

本篇完~