在上一章中，我們建立了一個工作佇列，工作佇列模式的設想是每一條訊息只會被轉發給一個消費者。本章將會講解完全不一樣的場景： 我們會把一個訊息轉發給多個消費者，這種模式稱之為釋出-訂閱模式。

　　為了闡述這個模式，我們將會搭建一個簡單的日誌系統，它包含兩種程式：一種傳送日誌訊息，另一種接收並列印日誌訊息。在這個日誌系統裡，每一個執行的消費者都可以獲取到訊息，在這種情況下，我們可以實現這種需求：一個消費者接收訊息並寫入磁碟，另一個消費者接收訊息並列印在電腦螢幕上。簡單來說，生產者釋出的訊息將會以廣播的形式轉發到所有的消費者。

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 
 8     private static final String EXCHANGE_NAME = "logs";
 9 
10     public static void main(String[] args) throws Exception {
11 
12         ConnectionFactory factory = new ConnectionFactory();
13         factory.setHost("192.168.92.130");
14 
15         try (Connection connection = factory.newConnection();
16              Channel channel = connection.createChannel();) {
17 
18             channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
19 
20             String message = "RabbitMq fanout。。。。。。";
21             channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes("utf-8"));
22 
23             System.out.println(" [x] Sent '" + message + "'");
24         }
25     }
26 }

　　正好你所看到的，Connection建立完成之後，定義了exchange，這一步是必要的，因為如果沒有交換器將無法傳送訊息。

　　如此沒有佇列繫結到該交換器上，那麼，交換器收到的訊息將會丟失，但是對我們本章的日誌系統來說沒問題的，當沒有消費者時，我們可以安全地放棄掉資料，我們只接收最新的日誌訊息。

　　ReceiveLogs.java

 1 public class ReceiveLogs {
 2 
 3     private static final String EXCHANGE_NAME = "logs";
 4 
 5     public static void main(String[] args) throws Exception {
 6 
 7         ConnectionFactory factory = new ConnectionFactory();
 8         factory.setHost("192.168.92.130");
 9 
10         Connection connection = factory.newConnection();
11         Channel channel = connection.createChannel();
12 
13         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
14 
15         final String queue = channel.queueDeclare().getQueue();
16         channel.queueBind(queue,EXCHANGE_NAME,"");
17 
18         System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
19 
20         DeliverCallback deliverCallback = (consumerTa,delivery) -> {
21 
22             String message = new String(delivery.getBody(), "UTF-8");
23             System.out.println(" [x] Received '" + message + "'");
24 
25         };
26 
27         channel.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}")
10     private String exchange;
11 
12     @Autowired
13     private AmqpTemplate amqpTemplate;
14 
15     public void send(String msg) {
16         amqpTemplate.convertAndSend(exchange,"",msg);
17     }
18 }

　　EmitLogRunner.java

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

@Component
public class EmitLogRunner implements ApplicationRunner {

    @Autowired
    private EmitLog emitLog;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        System.out.println("生產者釋出訊息：" + msg);
        emitLog.send("RabbitMq fanout test message");
    }
}

二、消費者程式碼　　

　　ReceiveInfoLogs.java

@Component
@RabbitListener(
        bindings = @QueueBinding(
                value = @Queue(value = "${rqbbitmq.log.fanout.info}",autoDelete = "true"),
                exchange = @Exchange(value = "${spring.rabbitmq.exchange}",type = ExchangeTypes.FANOUT)
        )
)
public class ReceiveInfoLogs {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @RabbitHandler
    public void receiveInfoLog (Object message) {
 
        System.out.println("接收到info級別的日誌：" + message);
    }
}

　　ReceiveErrorLogs.java

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(
        bindings = @QueueBinding(
                value = @Queue(value = "${rqbbitmq.log.fanout.error}",autoDelete = "true"),
                exchange = @Exchange(value = "${spring.rabbitmq.exchange}",type = ExchangeTypes.FANOUT)
        )
)
public class ReceiveErrorLogs {

    @Autowired
    private AmqpTemplate amqpTemplate;

    @RabbitHandler
    public void receiveErrorLog(Object message) {
        System.out.println("接收到的error級別日誌：" + message);
    }
}

　　注意看一下註解方式bindings裡面都是以@開頭並加上對應的要繫結的項，琢磨一下應該都能理解。

三、驗證

　啟動消費者和生產者，檢視控制檯：

點關注，不迷路，這是一個程式設計師都想要關注的公眾號

相關推薦