RabbitMQ指南之三：釋出/訂閱模式（Publish/Subscribe）
在上一章中，我們建立了一個工作佇列，工作佇列模式的設想是每一條訊息只會被轉發給一個消費者。本章將會講解完全不一樣的場景： 我們會把一個訊息轉發給多個消費者，這種模式稱之為釋出-訂閱模式。
為了闡述這個模式，我們將會搭建一個簡單的日誌系統，它包含兩種程式：一種傳送日誌訊息，另一種接收並列印日誌訊息。在這個日誌系統裡，每一個執行的消費者都可以獲取到訊息，在這種情況下，我們可以實現這種需求：一個消費者接收訊息並寫入磁碟，另一個消費者接收訊息並列印在電腦螢幕上。簡單來說，生產者釋出的訊息將會以廣播的形式轉發到所有的消費者。
1、交換器（Exchange）
在前兩章節我們，我們往佇列中釋出訊息或獲取訊息，然而，前面的講解其實並不完整，接下來，是時候介紹完整的RabbitMq訊息模型了。
回憶一下我們前兩章指南中包含的內容：
-
- 一個生產者用以傳送訊息；
- 一個佇列快取訊息；
- 一個消費者用以消費佇列中的訊息。
RabbitMq訊息模式的核心思想是：一個生產者並不會直接往一個佇列中傳送訊息，事實上，生產者根本不知道它傳送的訊息將被轉發到哪些佇列。
實際上，生產者只能把訊息傳送給一個exchange，exchange只做一件簡單的事情：一方面它們接收從生產者傳送過來的訊息，另一方面，它們把接收到的訊息推送給佇列。一個exchage必須清楚地知道如何處理一條訊息。
有四種類型的交換器，分別是：direct、topic、headers、fanout。本章主要講解最後一種：fanous（廣播模式）。下面建立一個fanout型別的交換器，我們稱之為:logs:
1 channel.exchangeDeclare("logs", "fanout");
廣播模式交換器很簡單，從字面意思也能理解，它其實就是把接收到的訊息推送給所有它知道的佇列。在我們的日誌系統中正好需要這種模式。
如果想檢視當前系統中有多少個exchange，可以使用以下命令：
sudo rabbitmqctl list_exchanges
或者通過控制檯檢視：
可以看到有很多以amq.*開頭的交換器，以及（AMQP default）預設交換器，這些是預設建立的交換器。
在前面兩章的指南中，我們並不知道交換器的存在，但是依然可以將訊息傳送到佇列中，那其實並不是因為我們可以不使用交換器，實際上是我們使用了預設的交換器（我們通過指定交換器為字字串：""），回顧一下我們之前是如何傳送訊息的：
1 channel.basicPublish("", "hello", null, message.getBytes());
第一個引數是交換器的名字，空字串表示它是一個預設或無命名的交換器，訊息將會由指定的路由鍵（第二個引數，routingKey，後面會講）轉發到佇列。
你可能會有疑問：既然exchange可以指定為空字串("")，那麼可否指定為null?
答案是：不能！
通過跟蹤釋出訊息的程式碼，在AMQImpl類中的Publish()方面中，可以看到，不光是exchange不能為null，同時routingKey路由鍵也不能為null，否則會丟擲異常：
接著上面的講解，我們建立一個命名的交換器：
1 channel.basicPublish( "logs", "", null, message.getBytes());
2、臨時佇列
在前兩章的例子中，我們使用的佇列都是有具體的佇列名，建立命名佇列是很必要的，因為我們需要將消費者指向同一名字的佇列。因此，要想在生產者和消費者中間共享佇列就必須要使用命名佇列。
但是，本章講解的日誌系統也可以使用非命名佇列（可以不手動命名），我們希望收到所有日誌訊息，而不是部分。並且我們希望總是接收到新的日誌訊息而不是舊的日誌訊息。為了解決這個問題，需要分兩步走。
首先，無論何時我們的消費者連線到RabbitMq，我們都需要一個新的、空的佇列來接收日誌訊息，因此，消費者在連線上RabbitMq之後需要建立一個任意名字的佇列，或者讓RabbitMq生成任意的佇列名字。
其次，一旦該消費者斷開了與RabbitMq的連線，佇列也被自動刪除。
通過JAVA客戶端的無參方法：queueDeclare()來建立一個 非持久化、專有的、自動刪除的、名字隨機生成 的佇列。
1 String queueName = channel.queueDeclare().getQueue();
3、繫結（Binding）
前面廣播模式的交換器和佇列已經建立好了，接下來就是要告訴交換器向佇列裡傳送訊息。交換器與佇列之間的關係稱之為繫結關係。
1 channel.queueBind(queueName, "logs", "");
至此，交換器已經可以往佇列中傳送訊息了。
可以通過下列命令來檢視佇列的繫結關係：
4、完整的程式碼
EmitLog.java
1 import com.rabbitmq.client.BuiltinExchangeType;
2 import com.rabbitmq.client.Channel;
3 import com.rabbitmq.client.Connection;
4 import com.rabbitmq.client.ConnectionFactory;
5
6 public class EmitLog {
7
8private static final String EXCHANGE_NAME = "logs";
9
10public static void main(String[] args) throws Exception {
11
12ConnectionFactory factory = new ConnectionFactory();
13factory.setHost("192.168.92.130");
14
15try (Connection connection = factory.newConnection();
16Channel channel = connection.createChannel();) {
17
18channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
19
20String message = "RabbitMq fanout。。。。。。";
21channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes("utf-8"));
22
23System.out.println(" [x] Sent '" + message + "'");
24}
25}
26 }
正好你所看到的，Connection建立完成之後，定義了exchange，這一步是必要的，因為如果沒有交換器將無法傳送訊息。
如此沒有佇列繫結到該交換器上，那麼，交換器收到的訊息將會丟失，但是對我們本章的日誌系統來說沒問題的，當沒有消費者時，我們可以安全地放棄掉資料，我們只接收最新的日誌訊息。
ReceiveLogs.java
1 public class ReceiveLogs {
2
3private static final String EXCHANGE_NAME = "logs";
4
5public static void main(String[] args) throws Exception {
6
7ConnectionFactory factory = new ConnectionFactory();
8factory.setHost("192.168.92.130");
9
10Connection connection = factory.newConnection();
11Channel channel = connection.createChannel();
12
13channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
14
15final String queue = channel.queueDeclare().getQueue();
16channel.queueBind(queue,EXCHANGE_NAME,"");
17
18System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
19
20DeliverCallback deliverCallback = (consumerTa,delivery) -> {
21
22String message = new String(delivery.getBody(), "UTF-8");
23System.out.println(" [x] Received '" + message + "'");
24
25};
26
27channel.basicConsume(queue,true,deliverCallback,consumerTag -> {});
28}
29 }
這裡的autoAck設定為true，因為我們這裡是廣播模式，每個消費者都會收到一樣的訊息，並且這裡給消費者生產的隨機名稱的佇列相當於是獨有的，所以在接收到訊息之後立即傳送確認回執是OK的。
但是這裡先提出一個疑問：在這種模式下，每個佇列收到的訊息是否也會有Ready和Unacked狀態？
5、測試結果
一、首先啟動生產者，再啟動兩個消費者
可以看到，生產者啟動後傳送的訊息丟失了，兩個消費者並沒有消費到，此時再看控制檯：
可見RabbitMq為我們建立了兩個隨機命名的佇列，其Exclusive是Owner，表示是專有的，Parameters為AD（auto delete），擁有該佇列的消費者一占斷開連線，佇列將會被自動刪除。
二、其次啟動生產者傳送一次訊息
兩個消費都都收到了訊息。
三、關閉所有消費者，觀察控制檯變化
兩個專有隨機佇列自動刪除了。
6、SpringBoot的實現
工程結構圖：
一、配置檔案application.properties：
生產者：
#RabbitMq spring.rabbitmq.host=192.168.92.130 spring.rabbitmq.exchange=logs
消費者：
#RabbitMq spring.rabbitmq.host=192.168.92.130 spring.rabbitmq.exchange=logs ##佇列--我們可以自己指定佇列名稱，也可以由RabbitMq自動生成，這裡為了方便，我們自己命名（如果需要，我也可以寫一個自動生成名稱的方法） rqbbitmq.log.fanout.info=info rqbbitmq.log.fanout.error=error server.port=8090
二、生產者程式碼
這裡為了讓系統生產者啟動時就自動傳送一條訊息，我加了一個EmitLogRunner類。
EmitLog.java
1 import org.springframework.amqp.core.AmqpTemplate;
2 import org.springframework.beans.factory.annotation.Autowired;
3 import org.springframework.beans.factory.annotation.Value;
4 import org.springframework.stereotype.Component;
5
6 @Component
7 public class EmitLog {
8
9@Value("${spring.rabbitmq.exchange}")
10private String exchange;
11
12@Autowired
13private AmqpTemplate amqpTemplate;
14
15public void send(String msg) {
16amqpTemplate.convertAndSend(exchange,"",msg);
17}
18 }
EmitLogRunner.java
1 import org.springframework.beans.factory.annotation.Autowired;
2 import org.springframework.boot.ApplicationArguments;
3 import org.springframework.boot.ApplicationRunner;
4 import org.springframework.stereotype.Component;
5
6 @Component
7 public class EmitLogRunner implements ApplicationRunner {
8
9@Autowired
10private EmitLog emitLog;
11
12@Override
13public void run(ApplicationArguments args) throws Exception {
14System.out.println("生產者釋出訊息：" + msg);
15emitLog.send("RabbitMq fanout test message");
16}
17 }
二、消費者程式碼
ReceiveInfoLogs.java
1 @Component
2 @RabbitListener(
3bindings = @QueueBinding(
4value = @Queue(value = "${rqbbitmq.log.fanout.info}",autoDelete = "true"),
5exchange = @Exchange(value = "${spring.rabbitmq.exchange}",type = ExchangeTypes.FANOUT)
6)
7 )
8 public class ReceiveInfoLogs {
9
10@Autowired
11private AmqpTemplate amqpTemplate;
12
13@RabbitHandler
14public void receiveInfoLog (Object message) {
15
16System.out.println("接收到info級別的日誌：" + message);
17}
18 }
ReceiveErrorLogs.java
1 import org.springframework.amqp.core.AmqpTemplate;
2 import org.springframework.amqp.core.ExchangeTypes;
3 import org.springframework.amqp.rabbit.annotation.*;
4 import org.springframework.beans.factory.annotation.Autowired;
5 import org.springframework.stereotype.Component;
6
7 @Component
8 @RabbitListener(
9bindings = @QueueBinding(
10value = @Queue(value = "${rqbbitmq.log.fanout.error}",autoDelete = "true"),
11exchange = @Exchange(value = "${spring.rabbitmq.exchange}",type = ExchangeTypes.FANOUT)
12)
13 )
14 public class ReceiveErrorLogs {
15
16@Autowired
17private AmqpTemplate amqpTemplate;
18
19@RabbitHandler
20public void receiveErrorLog(Object message) {
21System.out.println("接收到的error級別日誌：" + message);
22}
23 }
注意看一下註解方式bindings裡面都是以@開頭並加上對應的要繫結的項，琢磨一下應該都能理解。
三、驗證
啟動消費者和生產者，檢視控制檯：
至此，釋出訂閱模式講解完了，在下一章中將會講解Routing(路由)的概念。