譯: 3. RabbitMQ Spring AMQP 之 Publish/Subscribe 發佈和訂閱

在第一篇教程中，咱們展現瞭如何使用start.spring.io來利用Spring Initializr建立一個具備RabbitMQ starter dependency的項目來建立spring-amqp應用程序。

在上一個教程中，咱們建立了一個新的包（tut2）來放置咱們的配置，發送者和接收者，並建立了一個包含兩個使用者的工做隊列。工做隊列背後的假設是每一個任務都交付給一個工做者。

在這部分中，咱們將實現扇出模式，以向多個消費者傳遞消息。此模式稱爲 Publish/Subscribe 「發佈/訂閱」，並經過在Tut3Config文件中配置多個bean來實現。

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

Exchanges

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

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

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

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

相反，生產者只能向交易所發送消息。Exchanges交換是一件很是簡單的事情。一方面，它接收來自生產者的消息，另外一方面將它們推送到隊列。交易所必須確切知道如何處理收到的消息。它應該附加到特定隊列嗎？它應該附加到許多隊列嗎？或者它應該被丟棄。其規則由交換類型定義 。

 

 有幾種交換類型可供選擇：

• direct
• topic
• headers 
• fanout 

咱們將專一於最後一個 - fanout。讓咱們配置一個bean來描述這種類型的交換，並將其命名爲tut.fanout：

 Tut3Config.java

import org.springframework.amqp.core.AnonymousQueue;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;

import com.xingyun.springamqp.business.Tut3Receiver;
import com.xingyun.springamqp.business.Tut3Sender;

@Profile({ "tut3", "pub-sub", "publish-subscribe" })
@Configuration
public class Tut3Config {
    @Bean
    public FanoutExchange fanout() {
        return new FanoutExchange("tut.fanout");
    }

    @Profile("receiver")
    private static class ReceiverConfig {

        @Bean
        public Queue autoDeleteQueue1() {
            return new AnonymousQueue();
        }

        @Bean
        public Queue autoDeleteQueue2() {
            return new AnonymousQueue();
        }

        @Bean
        public Binding binding1(FanoutExchange fanout, Queue autoDeleteQueue1) {
            return BindingBuilder.bind(autoDeleteQueue1).to(fanout);
        }

        @Bean
        public Binding binding2(FanoutExchange fanout, Queue autoDeleteQueue2) {
            return BindingBuilder.bind(autoDeleteQueue2).to(fanout);
        }

        @Bean
        public Tut3Receiver receiver() {
            return new Tut3Receiver();
        }
    }

    @Profile("sender")
    @Bean
    public Tut3Sender sender() {
        return new Tut3Sender();
    }
}

 

咱們遵循與前兩個教程相同的方法。咱們建立了三個配置文件，即教程（「tut3」，「pub-sub」或「publish-subscribe」）。它們都是運行fanout 配置文件教程的同義詞。

接下來，咱們將FanoutExchange配置爲bean。

在「接收器」（Tut3Receiver）文件中，咱們定義「四個bean;

•    兩個autoDeleteQueues或AnonymousQueues
•    以及兩個綁定來將這些隊列綁定到交換機。

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

列出交換

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

sudo rabbitmqctl list_exchanges

 

在此列表中將有一些amq。*交換和默認（未命名）交換。這些是默認建立的，但目前您不太可能須要使用它們。

Nameless exchange 無名交換

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

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

 template.convertAndSend（fanout.getName（），「」，message）;

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

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

@Autowired
private RabbitTemplate template;

@Autowired
private FanoutExchange fanout;   // configured in Tut3Config above

template.convertAndSend(fanout.getName(), "", message);

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

臨時隊列

您可能還記得之前咱們使用過具備特定名稱的隊列（記住你好）。可以命名隊列對咱們來講相當重要 - 咱們須要將工做人員指向同一個隊列。

當您想要在生產者和消費者之間共享隊列時，爲隊列命名很重要。但咱們的粉絲示例並不是如此。

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

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

其次，一旦咱們斷開消費者，就應該自動刪除隊列。爲了使用spring-amqp客戶端，咱們定義了AnonymousQueue，它建立了一個帶有生成名稱的非持久的獨佔自動刪除隊列：

@Bean
public Queue autoDeleteQueue1() {
    return new AnonymousQueue();
}

@Bean
public Queue autoDeleteQueue2() {
    return new AnonymousQueue();
}

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

綁定

咱們已經建立了一個扇出交換和一個隊列。如今咱們須要告訴交換機將消息發送到咱們的隊列。交換和隊列之間的關係稱爲綁定。

在上面的Tut3Config中，您能夠看到咱們有兩個綁定，每一個AnonymousQueue一個。

@Bean
public Binding binding1(FanoutExchange fanout,
        Queue autoDeleteQueue1) {
    return BindingBuilder.bind(autoDeleteQueue1).to(fanout);
}

列出綁定

您可使用，您猜對了，列出現有綁定

rabbitmqctl list_bindings

把它們放在一塊兒

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

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

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

import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;

public class Tut3Sender {

    @Autowired
    private RabbitTemplate template;

    @Autowired
    private FanoutExchange fanout;

    int dots = 0;

    int count = 0;

    @Scheduled(fixedDelay = 1000, initialDelay = 500)
    public void send() {
        StringBuilder builder = new StringBuilder("Hello");
        if (dots++ == 3) {
            dots = 1;
        }
        for (int i = 0; i < dots; i++) {
            builder.append('.');
        }
        builder.append(Integer.toString(++count));
        String message = builder.toString();
        template.convertAndSend(fanout.getName(), "", message);
        System.out.println(" [x] Sent '" + message + "'");
    }
}

如您所見，咱們利用Tut3Config文件中的bean以及RabbitTemplate中的自動裝配以及咱們配置的FanoutExchange這一步是必要的，由於禁止發佈到不存在的交換。

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

消費者

Tut3Receiver.java

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.util.StopWatch;

public class Tut3Receiver {
    
    @RabbitListener(queues = "#{autoDeleteQueue1.name}")

    public void receive1(String in) throws InterruptedException {
        receive(in, 1);
    }

    @RabbitListener(queues = "#{autoDeleteQueue2.name}")
    public void receive2(String in) throws InterruptedException {
        receive(in, 2);
    }

    public void receive(String in, int receiver) throws InterruptedException {
        StopWatch watch = new StopWatch();
        watch.start();
        System.out.println("instance " + receiver + " [x] Received '" + in + "'");
        doWork(in);
        watch.stop();
        System.out.println("instance " + receiver + " [x] Done in " + watch.getTotalTimeSeconds() + "s");
    }

    private void doWork(String in) throws InterruptedException {
        for (char ch : in.toCharArray()) {
            if (ch == '.') {
                Thread.sleep(1000);
            }
        }
    }
}

 查看用法

java -jar RabbitMQ_0x03_SpringAMQP_PublishSubscribe_Sample-0.0.1-SNAPSHOT.jar

 

此次和以前有所不一樣，此次消費者和生產者必須同時運行才得行。

消費者和生產者等待時間都是60秒

啓動消費者

java -jar RabbitMQ_0x03_SpringAMQP_PublishSubscribe_Sample-0.0.1-SNAPSHOT.jar --spring.profiles.active=pub-sub,receiver

 顯示效果以下：

啓動生產者

java -jar RabbitMQ_0x03_SpringAMQP_PublishSubscribe_Sample-0.0.1-SNAPSHOT.jar --spring.profiles.active=pub-sub,sender

 顯示效果以下：