RabbitMQ实例教程：主题交换机

oi622

　　前面的例子中，尽管我们使用了direct路由代替fanout路由解决了盲目广播的问题，但direct路由也有它的缺陷，他不能基于多个标准做路由转发。

　　在上面的日志系统中，如果不仅想基于日志等级做订阅，也想根据日志的发生源做订阅该怎么处理呢？这时候你可能想到了unix系统工具中的syslog服务，它不仅基于日志等级（info/warn/crit...）进行路由转发，也会根据操作（auth/cron/kern...）做路由转发。

　　如果是那样的话，日志系统就灵活多了，它不仅能够监听来自‘cron’的关键错误，也能监听来自'kern'的所有日志。其实主题交换机（topic exchange）就能解决这种问题。

　　主题交换机（Topic exchange）

　　主题交换机的路由代码不能是任意写的，必须是小树点分隔开的一组单词列表。这些单词可以随便写，但通常是与连接消息特征有关的单词。有效地路由代码应该是这样的“stock.usd.nyse”，“nyse.vmw”，“quick.orange.rabbit”。路由代码可以随便写，但是长度限制在255字节。

　　注意，绑定代码也必须在同一个表单中。topic交换机与direct交换机类似-具有特定路由代码的消息会传送给所有匹配绑定代码的队列，但有两个特殊的绑定代码：

　　* ：它能替代一个单词　

　　#：它能替代0或多个单词

　　该例子中，我们给所有的动物发送消息，符合由三个单词（第一个单词描述速度；第二个单词描述颜色；第三个单词描述物种）组成的路由代码将会发送消息:“<speed>.<colour>.<species>”。

　　我们创建了三个绑定：Q1使用“*.orange.*”绑定，Q2使用“*.*.rabbit”和“lazy.#”绑定。这些绑定的意义如下：

　　Q1描述了所有颜色为橙色的动物。

　　Q2描述了是兔子的动物和懒惰的动物。

　　这样，“quick.orange.rabbit”消息通过路由转发给Q1、Q2两个队列。"lazy.orange.elephant"消息也会转发给Q1、Q2两个队列。“quick.orange.fox”消息只会转发给Q1队列，"lazy.brown.fox"也只会转发给Q2队列。"lazy.pink.rabbit"会转发给Q2队列一次，尽管它匹配两个绑定。"quick.brown.fox"并不匹配任何一个队列就会被废弃。

　　如果我们打破规则，每次只发一个或四个单词的话，如“orange”或”quick.orange.male.rabbit“，这些消息不匹配任何绑定，就会被废弃。但如果发送”lazy.orange.male.rabbit“这样的消息的话，由于它匹配最后的绑定仍会被转发到Q2队列中。

　　主题交换机是一种非常强大的交换机，当它只绑定”#“时，它会接收所有的消息，与fanout交换机类似。当没有使用”*“和”#“符号时，主题交换机的作用等同与direct交换机。

　　源代码

EmitLogTopic.java

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44

package com.favccxx.favrabbit; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class EmitLogTopic {     private static final String EXCHANGE_NAME = "topic_logs";     public static void main(String[] argv) {         Connection connection = null;         Channel channel = null;         try {             ConnectionFactory factory = new ConnectionFactory();             factory.setHost("localhost");             connection = factory.newConnection();             channel = connection.createChannel();             channel.exchangeDeclare(EXCHANGE_NAME, "topic");             String[] routingKeys = { "fast.orange.duck", "slow.orange.fish", "grey.rabbit", "fast.black.rabbit",                     "quick.white.rabbit", "lazy.dog", "lazy.black.pig" };             String[] messages = { "Hello", "Guys", "Girls", "Babies" };             for (int i = 0; i < routingKeys.length; i++) {                 for (int j = 0; j < messages.length; j++) {                     channel.basicPublish(EXCHANGE_NAME, routingKeys, null, messages[j].getBytes("UTF-8"));                     System.out.println(" [x] Sent '" + routingKeys + "':'" + messages[j] + "'");                 }             }         } catch (Exception e) {             e.printStackTrace();         } finally {             if (connection != null) {                 try {                     connection.close();                 } catch (Exception ignore) {                 }             }         }     } }

ReceiveLogsTopic.java

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41

package com.favccxx.favrabbit; import java.io.IOException; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.Consumer; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; public class ReceiveLogsTopic {     private static final String EXCHANGE_NAME = "topic_logs";     public static void main(String[] argv) throws Exception {         ConnectionFactory factory = new ConnectionFactory();         factory.setHost("localhost");         Connection connection = factory.newConnection();         Channel channel = connection.createChannel();         channel.exchangeDeclare(EXCHANGE_NAME, "topic");         String queueName = channel.queueDeclare().getQueue();         String[] bindingKeys = { "*.orange.*", "*.*.rabbit", "lazy.#" };         for (final String bindingKey : bindingKeys) {             channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);             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("[" + bindingKey + "] Received message :'" + message + "' from routingKey : " + envelope.getRoutingKey());                 }             };             channel.basicConsume(queueName, true, consumer);         }     } }

运行消息发送器，在消息接收平台输出内容如下

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

[*.orange.*] Received message :'Hello' from routingKey : fast.orange.duck [*.*.rabbit] Received message :'Guys' from routingKey : fast.orange.duck [lazy.#] Received message :'Girls' from routingKey : fast.orange.duck [*.orange.*] Received message :'Babies' from routingKey : fast.orange.duck [*.*.rabbit] Received message :'Hello' from routingKey : slow.orange.fish [lazy.#] Received message :'Guys' from routingKey : slow.orange.fish [*.orange.*] Received message :'Girls' from routingKey : slow.orange.fish [*.*.rabbit] Received message :'Babies' from routingKey : slow.orange.fish [lazy.#] Received message :'Hello' from routingKey : fast.black.rabbit [*.orange.*] Received message :'Guys' from routingKey : fast.black.rabbit [*.*.rabbit] Received message :'Girls' from routingKey : fast.black.rabbit [lazy.#] Received message :'Babies' from routingKey : fast.black.rabbit [*.orange.*] Received message :'Hello' from routingKey : quick.white.rabbit [*.*.rabbit] Received message :'Guys' from routingKey : quick.white.rabbit [lazy.#] Received message :'Girls' from routingKey : quick.white.rabbit [*.orange.*] Received message :'Babies' from routingKey : quick.white.rabbit [*.*.rabbit] Received message :'Hello' from routingKey : lazy.dog [lazy.#] Received message :'Guys' from routingKey : lazy.dog [*.orange.*] Received message :'Girls' from routingKey : lazy.dog [*.*.rabbit] Received message :'Babies' from routingKey : lazy.dog [lazy.#] Received message :'Hello' from routingKey : lazy.black.pig [*.orange.*] Received message :'Guys' from routingKey : lazy.black.pig [*.*.rabbit] Received message :'Girls' from routingKey : lazy.black.pig [lazy.#] Received message :'Babies' from routingKey : lazy.black.pig