RabbitMQ指南之三：發佈/訂閱模式（Publish/Subscribe）

在上一章中，咱們建立了一個工做隊列，工做隊列模式的設想是每一條消息只會被轉發給一個消費者。本章將會講解徹底不同的場景： 咱們會把一個消息轉發給多個消費者，這種模式稱之爲發佈-訂閱模式。

　　爲了闡述這個模式，咱們將會搭建一個簡單的日誌系統，它包含兩種程序：一種發送日誌消息，另外一種接收並打印日誌消息。在這個日誌系統裏，每個運行的消費者均可以獲取到消息，在這種狀況下，咱們能夠實現這種需求：一個消費者接收消息並寫入磁盤，另外一個消費者接收消息並打印在電腦屏幕上。簡單來講，生產者發佈的消息將會以廣播的形式轉發到全部的消費者。

一、交換器（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());

二、臨時隊列

　　在前兩章的例子中，咱們使用的隊列都是有具體的隊列名，建立命名隊列是很必要的，由於咱們須要將消費者指向同一名字的隊列。所以，要想在生產者和消費者中間共享隊列就必需要使用命名隊列。

　　可是，本章講解的日誌系統也可使用非命名隊列（能夠不手動命名），咱們但願收到全部日誌消息，而不是部分。而且咱們但願老是接收到新的日誌消息而不是舊的日誌消息。爲了解決這個問題，須要分兩步走。

　　首先，不管什麼時候咱們的消費者鏈接到RabbitMq，咱們都須要一個新的、空的隊列來接收日誌消息，所以，消費者在鏈接上RabbitMq以後須要建立一個任意名字的隊列，或者讓RabbitMq生成任意的隊列名字。

　　其次，一旦該消費者斷開了與RabbitMq的鏈接，隊列也被自動刪除。

　　經過JAVA客戶端的無參方法：queueDeclare()來建立一個非持久化、專有的、自動刪除的、名字隨機生成的隊列。

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

三、綁定（Binding）

　　前面廣播模式的交換器和隊列已經建立好了，接下來就是要告訴交換器向隊列裏發送消息。交換器與隊列之間的關係稱之爲綁定關係。

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

　　至此，交換器已經能夠往隊列中發送消息了。

　　能夠經過下列命令來查看隊列的綁定關係：

四、完整的代碼

　　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狀態？

五、測試結果

　　1、首先啓動生產者，再啓動兩個消費者

　　能夠看到，生產者啓動後發送的消息丟失了，兩個消費者並無消費到，此時再看控制檯：

　　可見RabbitMq爲咱們建立了兩個隨機命名的隊列，其Exclusive是Owner，表示是專有的，Parameters爲AD（auto delete），擁有該隊列的消費者一占斷開鏈接，隊列將會被自動刪除。

　　2、其次啓動生產者發送一次消息

　　兩個消費都都收到了消息。

　　3、關閉全部消費者，觀察控制檯變化

　　兩個專有隨機隊列自動刪除了。

六、SpringBoot的實現

　　工程結構圖：

1、配置文件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

2、生產者代碼

　　這裏爲了讓系統生產者啓動時就自動發送一條消息，我加了一個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");
    }
}

2、消費者代碼　　

　　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裏面都是以@開頭並加上對應的要綁定的項，琢磨一下應該都能理解。

3、驗證

　啓動消費者和生產者，查看控制檯：

點關注，不迷路，這是一個程序員都想要關注的公衆號