首页 > 代码库 > RabbitMQ 路由选择 (Routing)
RabbitMQ 路由选择 (Routing)
让日志接收者能够订阅部分消息。例如，我们可以仅仅将致命的错误写入日志文件，然而仍然在控制面板上打印出所有的其他类型的日志消息。
1、绑定（Bindings）
在前面中我们已经使用过绑定。类似下面的代码：
1 channel.queueBind(queueName, EXCHANGE_NAME, "");
绑定表示转发器与队列之间的关系。我们也可以简单的认为：队列对该转发器上的消息感兴趣。
绑定可以附带一个额外的参数routingKey。为了与避免basicPublish方法（发布消息的方法）的参数混淆，我们准备把它称作绑定键（binding key）。下面展示如何使用绑定键（binding key）来创建一个绑定：
1 channel.queueBind(queueName, EXCHANGE_NAME, "black");
绑定键的意义依赖于转发器的类型。对于fanout类型，忽略此参数。
2、直接转发（Direct exchange）
日志系统广播所有的消息给所有的消费者。我们希望可以对其扩展，来允许根据日志的严重性进行过滤日志。例如：我们可能希望把致命类型的错误写入硬盘，而不把硬盘空间浪费在警告或者消息类型的日志上。
之前我们使用fanout类型的转发器，但是并没有给我们带来更多的灵活性：仅仅可以愚蠢的转发。
我们将会使用direct类型的转发器进行替代。direct类型的转发器背后的路由转发算法很简单：消息会被推送至绑定键（binding key）和消息发布附带的选择键（routing key）完全匹配的队列。
上图，我们可以看到direct类型的转发器与两个队列绑定。第一个队列与绑定键orange绑定，第二个队列与转发器间有两个绑定，一个与绑定键black绑定，另一个与green绑定键绑定。
这样的话，当一个消息附带一个选择键（routing key） orange发布至转发器将会被导向到队列Q1。消息附带一个选择键（routing key）black或者green将会被导向到Q2.所有的其他的消息将会被丢弃。
3、多重绑定（multiple bindings）
使用一个绑定键（binding key）绑定多个队列是完全合法的。如上图，一个附带选择键（routing key）的消息将会被转发到Q1和Q2。
4、发送日志（Emittinglogs）
我们准备将这种模式用于我们的日志系统。我们将消息发送到direct类型的转发器而不是fanout类型。我们将把日志的严重性作为选择键（routing key）。这样的话，接收程序可以根据严重性来选择接收。我们首先关注发送日志的代码：
像以前一样，我们需要先创建一个转发器：
1 channel.exchangeDeclare(EXCHANGE_NAME,"direct");
然后我们准备发送一条消息：
1 channel.basicPublish(EXCHANGE_NAME,severity, null, message.getBytes());
为了简化代码，我们假定‘severity’是‘info’，‘warning’，‘error’中的一个。
5、订阅
接收消息的代码和前面的博客的中类似，只有一点不同：我们给我们所感兴趣的严重性类型的日志创建一个绑定。
1 StringqueueName = channel.queueDeclare().getQueue(); 2 3 for(Stringseverity : argv) 4 5 { 6 7 channel.queueBind(queueName, EXCHANGE_NAME, severity); 8 9 }
6、完整的实例
发送端
1 public class EmitLogDirect 2 { 3 4 private static final String EXCHANGE_NAME = "ex_logs_direct"; 5 private static final String[] SEVERITIES = { "info", "warning", "error" }; 6 7 public static void main(String[] argv) throws java.io.IOException 8 { 9 // 创建连接和频道 10 ConnectionFactory factory = new ConnectionFactory(); 11 factory.setHost("localhost"); 12 Connection connection = factory.newConnection(); 13 Channel channel = connection.createChannel(); 14 // 声明转发器的类型 15 channel.exchangeDeclare(EXCHANGE_NAME, "direct"); 16 17 //发送6条消息 18 for (int i = 0; i < 6; i++) 19 { 20 String severity = getSeverity(); 21 String message = severity + "_log :" + UUID.randomUUID().toString(); 22 // 发布消息至转发器，指定routingkey 23 channel.basicPublish(EXCHANGE_NAME, severity, null, message 24 .getBytes()); 25 System.out.println(" [x] Sent ‘" + message + "‘"); 26 } 27 28 channel.close(); 29 connection.close(); 30 } 31 32 /** 33 * 随机产生一种日志类型 34 * 35 * @return 36 */ 37 private static String getSeverity() 38 { 39 Random random = new Random(); 40 int ranVal = random.nextInt(3); 41 return SEVERITIES[ranVal]; 42 } 43 }
接收端：
1 public class ReceiveLogsDirect 2 { 3 4 private static final String EXCHANGE_NAME = "ex_logs_direct"; 5 private static final String[] SEVERITIES = { "info", "warning", "error" }; 6 7 public static void main(String[] argv) throws java.io.IOException, 8 java.lang.InterruptedException 9 { 10 // 创建连接和频道 11 ConnectionFactory factory = new ConnectionFactory(); 12 factory.setHost("localhost"); 13 Connection connection = factory.newConnection(); 14 Channel channel = connection.createChannel(); 15 // 声明direct类型转发器 16 channel.exchangeDeclare(EXCHANGE_NAME, "direct"); 17 18 String queueName = channel.queueDeclare().getQueue(); 19 String severity = getSeverity(); 20 // 指定binding_key 21 channel.queueBind(queueName, EXCHANGE_NAME, severity); 22 System.out.println(" [*] Waiting for "+severity+" logs. 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 32 System.out.println(" [x] Received ‘" + message + "‘"); 33 } 34 } 35 36 /** 37 * 随机产生一种日志类型 38 * 39 * @return 40 */ 41 private static String getSeverity() 42 { 43 Random random = new Random(); 44 int ranVal = random.nextInt(3); 45 return SEVERITIES[ranVal]; 46 } 47 }
1 运行结果： 2 [x] Sent ‘error_log :d142b096-46c0-4380-a1d2-d8b2ac136a9c‘ 3 [x] Sent ‘error_log :55ee1fc4-c87c-4e5e-81ba-49433890b9ce‘ 4 [x] Sent ‘error_log :d01877d6-87c7-4e0a-a109-697d122bc4c9‘ 5 [x] Sent ‘error_log :b42471b1-875c-43f1-b1ea-0dd5b49863f3‘ 6 [x] Sent ‘info_log :a6c1bc87-efb0-43eb-8314-8a74c345ed05‘ 7 [x] Sent ‘info_log :b6a84b6a-353e-4e88-8c23-c791d93b44be‘ 8 ------------------------------------------------------------------------------------ 9 [*] Waiting for error logs. To exit press CTRL+C 10 [x] Received ‘error_log :d142b096-46c0-4380-a1d2-d8b2ac136a9c‘ 11 [x] Received ‘error_log :55ee1fc4-c87c-4e5e-81ba-49433890b9ce‘ 12 [x] Received ‘error_log :d01877d6-87c7-4e0a-a109-697d122bc4c9‘ 13 [x] Received ‘error_log :b42471b1-875c-43f1-b1ea-0dd5b49863f3‘ 14 ------------------------------------------------------------------------------------ 15 [*] Waiting for error logs. To exit press CTRL+C 16 [x] Received ‘error_log :d142b096-46c0-4380-a1d2-d8b2ac136a9c‘ 17 [x] Received ‘error_log :55ee1fc4-c87c-4e5e-81ba-49433890b9ce‘ 18 [x] Received ‘error_log :d01877d6-87c7-4e0a-a109-697d122bc4c9‘ 19 [x] Received ‘error_log :b42471b1-875c-43f1-b1ea-0dd5b49863f3‘ 20 ------------------------------------------------------------------------------------ 21 [*] Waiting for info logs. To exit press CTRL+C 22 [x] Received ‘info_log :a6c1bc87-efb0-43eb-8314-8a74c345ed05‘ 23 [x] Received ‘info_log :b6a84b6a-353e-4e88-8c23-c791d93b44be‘ 24 25 可以看到我们实现了博文开头所描述的特性，接收者可以自定义自己感兴趣类型的日志。
其实文章这么长就在说：发送消息时可以设置routing_key，接收队列与转发器间可以设置binding_key，接收者接收与binding_key与routing_key相同的消息。
参考博客：http://blog.csdn.net/lmj623565791/article/details/37669573
RabbitMQ 路由选择 (Routing)