1.5 RabbitMQ 實戰，主題 · java核心知識整理

## 主題交換（Topic exchange）使用 topic 類型的交換器，不能有任意的綁定鍵，它必須是由點隔開的一系列的標識符組成。標識符可以是任何東西，但通常它們指定與消息相關聯的一些功能。其中，有幾個有效的綁定鍵，例如 “stock.usd.nyse”， “nyse.vmw”， “quick.orange.rabbit”。可以有任何數量的標識符，最多可達 255 個字節。 topic 類型的交換器和 direct 類型的交換器很類似，一個特定路由的消息將被傳遞到與匹配的綁定鍵綁定的匹配的所有隊列。關于綁定鍵有兩種有兩個重要的特殊情況： ~~~null * 可以匹配一個標識符。 # 可以匹配零個或多個標識符。 ~~~ [![](https://gitee.com/chenssy/blog-home/raw/master/image/201810/rabbitmq_python-five.png)](https://gitee.com/chenssy/blog-home/raw/master/image/201810/rabbitmq_python-five.png) 在這個例子中，我們將發送所有描述動物的消息。消息將使用由三個字（兩個點）組成的綁定鍵發送。綁定鍵中的第一個字將描述速度，第二個顏色和第三個種類：“..”。其中， Q1 對所有的橙色動物感興趣。而 Q2 想聽聽有關兔子的一切，以及關于懶惰動物的一切。如果我們違反合同并發送一個或四個字的消息，如 “quick.orange.male.rabbit” 會發生什么？那么，這些消息將不會匹配任何綁定，并將被丟失。 topic 類型的交換器是強大的，可以實現其他類型的交換器。當一個隊列與“＃”綁定綁定鍵時，它將接收所有消息，類似 fanout 類型的交換器。當一個隊列與`“*”`和`“＃”`在綁定中不被使用時，類似 direct 類型的交換器。 ## 案例實戰 ### 發送端發送端，連接到 RabbitMQ，發送一條數據，然后退出。 ~~~java public class EmitLogTopic { private static final String EXCHANGE_NAME = "topic_logs"; private static final String[] LOG_LEVEL_ARR = {"dao.debug", "dao.info", "dao.error", "service.debug", "service.info", "service.error", "controller.debug", "controller.info", "controller.error"}; public static void main(String[] args) throws IOException, TimeoutException { // 創建連接 ConnectionFactory factory = new ConnectionFactory(); // 設置 RabbitMQ 的主機名 factory.setHost("localhost"); // 創建一個連接 Connection connection = factory.newConnection(); // 創建一個通道 Channel channel = connection.createChannel(); // 指定一個交換器 channel.exchangeDeclare(EXCHANGE_NAME, "topic"); // 發送消息 for (String severity : LOG_LEVEL_ARR) { String message = "Liang-MSG log : [" +severity+ "]" + UUID.randomUUID().toString(); // 發布消息至交換器 channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); } // 關閉頻道和連接 channel.close(); connection.close(); } } ~~~ ### 接受端接受端，不斷等待服務器推送消息，然后在控制臺輸出。 ~~~java public class ReceiveLogsTopic { private static final String EXCHANGE_NAME = "topic_logs"; private static final String[] LOG_LEVEL_ARR = {"#", "dao.error", "*.error", "dao.*", "service.#", "*.controller.#"}; public static void main(String[] args) throws IOException, TimeoutException { // 創建連接 ConnectionFactory factory = new ConnectionFactory(); // 設置 RabbitMQ 的主機名 factory.setHost("localhost"); // 創建一個連接 Connection connection = factory.newConnection(); // 創建一個通道 Channel channel = connection.createChannel(); // 指定一個交換器 channel.exchangeDeclare(EXCHANGE_NAME, "topic"); // 設置日志級別 int rand = new Random().nextInt(5); String severity = LOG_LEVEL_ARR[rand]; // 創建一個非持久的、唯一的、自動刪除的隊列 String queueName = channel.queueDeclare().getQueue(); // 綁定交換器和隊列 channel.queueBind(queueName, EXCHANGE_NAME, severity); // 打印 System.out.println(" [*] LOG INFO : " + severity); // 創建隊列消費者 final 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); } } ~~~ 現在，做一個實驗，我們開啟三個 ReceiveLogsTopic 工作程序：ReceiveLogsTopic1、 ReceiveLogsTopic2 與 ReceiveLogsTopic3。 ReceiveLogsTopic1 ~~~null [*] LOG INFO : dao.error [x] Received 'Liang-MSG log : [dao.error]041cd8ba-df7d-4d20-a11f-ba21a0c2a02a' ~~~ ReceiveLogsTopic2 ~~~null [*] LOG INFO : *.error [x] Received 'Liang-MSG log : [dao.error]041cd8ba-df7d-4d20-a11f-ba21a0c2a02a' [x] Received 'Liang-MSG log : [service.error]e3565f12-9782-4c22-a91c-f513f31b037d' [x] Received 'Liang-MSG log : [controller.error]4436101a-3346-41f6-a9af-b8a4fbda451e' ~~~ ReceiveLogsTopic3 ~~~null [*] LOG INFO : # [x] Received 'Liang-MSG log : [dao.debug]4eb08245-2c05-490b-a5a5-2742cb70d831' [x] Received 'Liang-MSG log : [dao.info]e9d4073b-1e61-4c6f-b531-ac42eaa346af' [x] Received 'Liang-MSG log : [dao.error]041cd8ba-df7d-4d20-a11f-ba21a0c2a02a' [x] Received 'Liang-MSG log : [service.debug]0ec84cbf-47ab-4813-a5db-e57d5e78830e' [x] Received 'Liang-MSG log : [service.info]2e12e1b7-7a09-4eb7-8ad1-8e53f533121c' [x] Received 'Liang-MSG log : [service.error]e3565f12-9782-4c22-a91c-f513f31b037d' [x] Received 'Liang-MSG log : [controller.debug]94e5be72-15f6-496d-84f3-2a107bafc92b' [x] Received 'Liang-MSG log : [controller.info]62bbe378-617d-4214-beb4-98cc53e73272' [x] Received 'Liang-MSG log : [controller.error]4436101a-3346-41f6-a9af-b8a4fbda451e' ~~~ 此時，ReceiveLogsTopic1 、ReceiveLogsTopic2 與 ReceiveLogsTopic3 同時收到了屬于自己級別的消息。我們發現，ReceiveLogsTopic1、ReceiveLogsTopic2、ReceiveLogsTopic3、ReceiveLogsTopic4同時收到了屬于自己匹配的消息。尤其是ReceiveLogsTopic1 類似于 direct 類型的交換器，ReceiveLogsTopic3 類似于 fanout 類型的交換器。 ## 源代碼 > 相關示例完整代碼：[https://github.com/lianggzone/rabbitmq-action](https://github.com/lianggzone/rabbitmq-action)