譯: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 是隨機生成的隊列名稱。 

本篇完~