我們進(jìn)步改良了我們的日志系統(tǒng)。我們使用direct類型轉(zhuǎn)發(fā)器，使得接收者有能力進(jìn)行選擇性的接收日志,，而非fanout那樣，只能夠無腦的轉(zhuǎn)發(fā)。

　　雖然使用direct類型改良了我們的系統(tǒng)，但是仍然存在一些局限性：它不能夠基于多重條件進(jìn)行路由選擇。
　　在我們的日志系統(tǒng)中，我們有可能希望不僅根據(jù)日志的級別而且想根據(jù)日志的來源進(jìn)行訂閱。這個概念類似unix工具：syslog，它轉(zhuǎn)發(fā)日志基于嚴(yán)重性（info/warning/crit…）和設(shè)備（auth/cron/kern…）
　　這樣可能給我們更多的靈活性：我們可能只想訂閱來自’cron’的致命錯誤日志，而不是來自’kern’的。
　　為了在我們的系統(tǒng)中實現(xiàn)上述的需求，我們需要學(xué)習(xí)稍微復(fù)雜的主題類型的轉(zhuǎn)發(fā)器（topic exchange）。

　　1、 主題轉(zhuǎn)發(fā)（Topic Exchange）

　　發(fā)往主題類型的轉(zhuǎn)發(fā)器的消息不能隨意的設(shè)置選擇鍵（routing_key），必須是由點隔開的一系列的標(biāo)識符組成。標(biāo)識符可以是任何東西，但是一般都與消息的某些特性相關(guān)。一些合法的選擇鍵的例子："stock.usd.nyse", "nyse.vmw","quick.orange.rabbit".你可以定義任何數(shù)量的標(biāo)識符，上限為255個字節(jié)。
　　綁定鍵和選擇鍵的形式一樣。主題類型的轉(zhuǎn)發(fā)器背后的邏輯和直接類型的轉(zhuǎn)發(fā)器很類似：一個附帶特殊的選擇鍵將會被轉(zhuǎn)發(fā)到綁定鍵與之匹配的隊列中。需要注意的是：關(guān)于綁定鍵有兩種特殊的情況。
　　*可以匹配一個標(biāo)識符。
　　#可以匹配0個或多個標(biāo)識符

　　2、 圖解：

?　　我們準(zhǔn)備發(fā)送關(guān)于動物的消息。消息會附加一個選擇鍵包含3個標(biāo)識符（兩個點隔開）。第一個標(biāo)識符描述動物的速度，第二個標(biāo)識符描述動物的顏色，第三個標(biāo)識符描述動物的物種：<speed>.<color>.<species>。
　　我們創(chuàng)建3個綁定鍵：Q1與*.orange.*綁定Q2與*.*.rabbit和lazy.#綁定。
　　可以簡單的認(rèn)為:
　　Q1對所有的橙色動物感興趣。
　　Q2想要知道關(guān)于兔子的一切以及關(guān)于懶洋洋的動物的一切。
　　一個附帶quick.orange.rabbit的選擇鍵的消息將會被轉(zhuǎn)發(fā)到兩個隊列。附帶lazy.orange.elephant的消息也會被轉(zhuǎn)發(fā)到兩個隊列。另一方面quick.orange.fox只會被轉(zhuǎn)發(fā)到Q1，lazy.brown.fox將會被轉(zhuǎn)發(fā)到Q2。lazy.pink.rabbit雖然與兩個綁定鍵匹配，但是也只會被轉(zhuǎn)發(fā)到Q2一次。quick.brown.fox不能與任何綁定鍵匹配，所以會被丟棄。
　　如果我們違法我們的約定，發(fā)送一個或者四個標(biāo)識符的選擇鍵，類似：orange，quick.orange.male.rabbit，這些選擇鍵不能與任何綁定鍵匹配，所以消息將會被丟棄。
　　另一方面，lazy.orange.male.rabbit，雖然是四個標(biāo)識符，也可以與lazy.#匹配，從而轉(zhuǎn)發(fā)至Q2。
　　注：主題類型的轉(zhuǎn)發(fā)器非常強大，可以實現(xiàn)其他類型的轉(zhuǎn)發(fā)器。
　　當(dāng)一個隊列與綁定鍵#綁定，將會收到所有的消息，類似fanout類型轉(zhuǎn)發(fā)器。
　　當(dāng)綁定鍵中不包含任何#與*時，類似direct類型轉(zhuǎn)發(fā)器。

　　3、 完整的例子

　　發(fā)送端

1 public class EmitLogTopic 2 { 3 4 private static final String EXCHANGE_NAME = "topic_logs"; 5 6 public static void main(String[] argv) throws Exception 7 { 8 // 創(chuàng)建連接和頻道 9 ConnectionFactory factory = new ConnectionFactory(); 10 factory.setHost("localhost"); 11 Connection connection = factory.newConnection(); 12 Channel channel = connection.createChannel(); 13 14 channel.exchangeDeclare(EXCHANGE_NAME, "topic"); 15 16 String[] routing_keys = new String[] { "kernal.info", "cron.warning", 17 "auth.info", "kernel.critical" }; 18 for (String routing_key : routing_keys) 19 { 20 String msg = UUID.randomUUID().toString(); 21 channel.basicPublish(EXCHANGE_NAME, routing_key, null, msg 22 .getBytes()); 23 System.out.println(" [x] Sent routingKey = "+routing_key+" ,msg = " + msg + "."); 24 } 25 26 channel.close(); 27 connection.close(); 28 } 29 }

　　接收端1：

1 public class ReceiveLogsTopicForKernel 2 { 3 4 private static final String EXCHANGE_NAME = "topic_logs"; 5 6 public static void main(String[] argv) throws Exception 7 { 8 // 創(chuàng)建連接和頻道 9 ConnectionFactory factory = new ConnectionFactory(); 10 factory.setHost("localhost"); 11 Connection connection = factory.newConnection(); 12 Channel channel = connection.createChannel(); 13 // 聲明轉(zhuǎn)發(fā)器 14 channel.exchangeDeclare(EXCHANGE_NAME, "topic"); 15 // 隨機生成一個隊列 16 String queueName = channel.queueDeclare().getQueue(); 17 18 //接收所有與kernel相關(guān)的消息 19 channel.queueBind(queueName, EXCHANGE_NAME, "kernel.*"); 20 21 System.out.println(" [*] Waiting for messages about kernel. To exit press CTRL+C"); 22 23 QueueingConsumer consumer = new QueueingConsumer(channel); 24 channel.basicConsume(queueName, true, consumer); 25 26 while (true) 27 { 28 QueueingConsumer.Delivery delivery = consumer.nextDelivery(); 29 String message = new String(delivery.getBody()); 30 String routingKey = delivery.getEnvelope().getRoutingKey(); 31 32 System.out.println(" [x] Received routingKey = " + routingKey 33 + ",msg = " + message + "."); 34 } 35 } 36 }

　　接收端2：

1 public class ReceiveLogsTopicForCritical 2 { 3 4 private static final String EXCHANGE_NAME = "topic_logs"; 5 6 public static void main(String[] argv) throws Exception 7 { 8 // 創(chuàng)建連接和頻道 9 ConnectionFactory factory = new ConnectionFactory(); 10 factory.setHost("localhost"); 11 Connection connection = factory.newConnection(); 12 Channel channel = connection.createChannel(); 13 // 聲明轉(zhuǎn)發(fā)器 14 channel.exchangeDeclare(EXCHANGE_NAME, "topic"); 15 // 隨機生成一個隊列 16 String queueName = channel.queueDeclare().getQueue(); 17 18 // 接收所有與kernel相關(guān)的消息 19 channel.queueBind(queueName, EXCHANGE_NAME, "*.critical"); 20 21 System.out 22 .println(" [*] Waiting for critical messages. To exit press CTRL+C"); 23 24 QueueingConsumer consumer = new QueueingConsumer(channel); 25 channel.basicConsume(queueName, true, consumer); 26 27 while (true) 28 { 29 QueueingConsumer.Delivery delivery = consumer.nextDelivery(); 30 String message = new String(delivery.getBody()); 31 String routingKey = delivery.getEnvelope().getRoutingKey(); 32 33 System.out.println(" [x] Received routingKey = " + routingKey 34 + ",msg = " + message + "."); 35 } 36 } 37 }

　　只接收致命錯誤的日志消息。

?　　運行結(jié)果：

?[x] Sent routingKey = kernal.info ,msg = a7261f0d-18cc-4c85-ba80-5ecd9283dae7.
?[x] Sent routingKey = cron.warning ,msg = 0c7e4484-66e0-4846-a869-a7a266e16281.
?[x] Sent routingKey = auth.info ,msg = 3273f21f-6e6e-42f2-83df-1f2fafa7a19a.
?[x] Sent routingKey = kernel.critical ,msg = f65d3e1a-0619-4f85-8b0d-59375380ecc9.

--------------------------------------------------------------------------------------------------------------------

?[*] Waiting for messages about kernel. To exit press CTRL+C
?[x] Received routingKey = kernel.critical,msg = f65d3e1a-0619-4f85-8b0d-59375380ecc9.

--------------------------------------------------------------------------------------------------------------------

?[*] Waiting for critical messages. To exit press CTRL+C
?[x] Received routingKey = kernel.critical,msg = f65d3e1a-0619-4f85-8b0d-59375380ecc9.

　　可以看到，我們通過使用topic類型的轉(zhuǎn)發(fā)器，成功實現(xiàn)了多重條件選擇的訂閱。

參考博客：http://blog.csdn.net/lmj623565791/article/details/37706355

與50位技術(shù)專家面對面20年技術(shù)見證，附贈技術(shù)全景圖

總結(jié)

以上是生活随笔為你收集整理的RabbitMQ 主题（Topic）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。