RabbitMQ （四）路由选择 (Routing)

电玩女神 2022-08-23 09:55 181阅读 0赞

上一篇博客我们建立了一个简单的日志系统，我们能够广播日志消息给所有你的接收者，如果你不了解，请查看：[RabbitMQ （三） 发布/订阅][RabbitMQ _]。本篇博客我们准备给日志系统添加新的特性，让日志接收者能够订阅部分消息。例如，我们可以仅仅将致命的错误写入日志文件，然而仍然在控制面板上打印出所有的其他类型的日志消息。

###### 1、绑定（Bindings） ######

在上一篇博客中我们已经使用过绑定。类似下面的代码：  
channel.queueBind(queueName, EXCHANGE\_NAME, "");  
绑定表示转发器与队列之间的关系。我们也可以简单的认为：队列对该转发器上的消息感兴趣。  
绑定可以附带一个额外的参数routingKey。为了与避免basicPublish方法（发布消息的方法）的参数混淆，我们准备把它称作绑定键（binding key）。下面展示如何使用绑定键（binding key）来创建一个绑定：  
channel.queueBind(queueName, EXCHANGE\_NAME, "black");  
绑定键的意义依赖于转发器的类型。对于fanout类型，忽略此参数。

###### 2、直接转发（Direct exchange） ######

上一篇的日志系统广播所有的消息给所有的消费者。我们希望可以对其扩展，来允许根据日志的严重性进行过滤日志。例如：我们可能希望把致命类型的错误写入硬盘，而不把硬盘空间浪费在警告或者消息类型的日志上。  
之前我们使用fanout类型的转发器，但是并没有给我们带来更多的灵活性：仅仅可以愚蠢的转发。  
我们将会使用direct类型的转发器进行替代。direct类型的转发器背后的路由转发算法很简单：消息会被推送至绑定键（binding key）和消息发布附带的选择键（routing key）完全匹配的队列。  
图解：

![Center][]

上图，我们可以看到direct类型的转发器与两个队列绑定。第一个队列与绑定键orange绑定，第二个队列与转发器间有两个绑定，一个与绑定键black绑定，另一个与green绑定键绑定。  
这样的话，当一个消息附带一个选择键（routing key） orange发布至转发器将会被导向到队列Q1。消息附带一个选择键（routing key）black或者green将会被导向到Q2.所有的其他的消息将会被丢弃。

###### 3、多重绑定（multiple bindings） ######

![Center 1][]

使用一个绑定键（binding key）绑定多个队列是完全合法的。如上图，一个附带选择键（routing key）的消息将会被转发到Q1和Q2。

###### 4、发送日志（Emittinglogs） ######

我们准备将这种模式用于我们的日志系统。我们将消息发送到direct类型的转发器而不是fanout类型。我们将把日志的严重性作为选择键（routing key）。这样的话，接收程序可以根据严重性来选择接收。我们首先关注发送日志的代码：

像以前一样，我们需要先创建一个转发器：

channel.exchangeDeclare(EXCHANGE\_NAME,"direct");

然后我们准备发送一条消息：

channel.basicPublish(EXCHANGE\_NAME,severity, null, message.getBytes());

为了简化代码，我们假定‘severity’是‘info’，‘warning’，‘error’中的一个。

###### 5、订阅 ######

接收消息的代码和前面的博客的中类似，只有一点不同：我们给我们所感兴趣的严重性类型的日志创建一个绑定。

StringqueueName = channel.queueDeclare().getQueue();

for(Stringseverity : argv)

channel.queueBind(queueName, EXCHANGE\_NAME, severity);

###### 6、完整的实例 ######

![Center 2][]

发送端：EmitLogDirect.java

**\[java\]**  [view plain][]   [copy][view plain]

1.  **package** com.zhy.rabbit.\_04\_binding\_key;  
2.    
3.  **import** java.util.Random;  
4.  **import** java.util.UUID;  
5.    
6.  **import** com.rabbitmq.client.Channel;  
7.  **import** com.rabbitmq.client.Connection;  
8.  **import** com.rabbitmq.client.ConnectionFactory;  
9.    
10. **public** **class** EmitLogDirect  
11. \{  
12.   
13.     **private** **static** **final** String EXCHANGE\_NAME = "ex\_logs\_direct";  
14.     **private** **static** **final** String\[\] SEVERITIES = \{ "info", "warning", "error" \};  
15.   
16.     **public** **static** **void** main(String\[\] argv) **throws** java.io.IOException  
17.     \{  
18.         // 创建连接和频道  
19.         ConnectionFactory factory = **new** ConnectionFactory();  
20.         factory.setHost("localhost");  
21.         Connection connection = factory.newConnection();  
22.         Channel channel = connection.createChannel();  
23.         // 声明转发器的类型  
24.         channel.exchangeDeclare(EXCHANGE\_NAME, "direct");  
25.   
26.         //发送6条消息  
27.         **for** (**int** i = 0; i < 6; i++)  
28.         \{  
29.             String severity = getSeverity();  
30.             String message = severity + "\_log :" + UUID.randomUUID().toString();  
31.             // 发布消息至转发器，指定routingkey  
32.             channel.basicPublish(EXCHANGE\_NAME, severity, **null**, message  
33.                     .getBytes());  
34.             System.out.println(" \[x\] Sent '" + message + "'");  
35.         \}  
36.   
37.         channel.close();  
38.         connection.close();  
39.     \}  
40.   
41.     /\*\* 
42.      \* 随机产生一种日志类型 
43.      \*  
44.      \* @return 
45.      \*/  
46.     **private** **static** String getSeverity()  
47.     \{  
48.         Random random = **new** Random();  
49.         **int** ranVal = random.nextInt(3);  
50.         **return** SEVERITIES\[ranVal\];  
51.     \}  
52. \}

随机发送6条随机类型（routing key）的日志给转发器~~

接收端：ReceiveLogsDirect.java

**\[java\]**  [view plain][]   [copy][view plain]

1.  **package** com.zhy.rabbit.\_04\_binding\_key;  
2.    
3.  **import** java.util.Random;  
4.    
5.  **import** com.rabbitmq.client.Channel;  
6.  **import** com.rabbitmq.client.Connection;  
7.  **import** com.rabbitmq.client.ConnectionFactory;  
8.  **import** com.rabbitmq.client.QueueingConsumer;  
9.    
10. **public** **class** ReceiveLogsDirect  
11. \{  
12.   
13.     **private** **static** **final** String EXCHANGE\_NAME = "ex\_logs\_direct";  
14.     **private** **static** **final** String\[\] SEVERITIES = \{ "info", "warning", "error" \};  
15.   
16.     **public** **static** **void** main(String\[\] argv) **throws** java.io.IOException,  
17.             java.lang.InterruptedException  
18.     \{  
19.         // 创建连接和频道  
20.         ConnectionFactory factory = **new** ConnectionFactory();  
21.         factory.setHost("localhost");  
22.         Connection connection = factory.newConnection();  
23.         Channel channel = connection.createChannel();  
24.         // 声明direct类型转发器  
25.         channel.exchangeDeclare(EXCHANGE\_NAME, "direct");  
26.   
27.         String queueName = channel.queueDeclare().getQueue();  
28.         String severity = getSeverity();  
29.         // 指定binding\_key  
30.         channel.queueBind(queueName, EXCHANGE\_NAME, severity);  
31.         System.out.println(" \[\*\] Waiting for "\+severity+" logs. To exit press CTRL+C");  
32.   
33.         QueueingConsumer consumer = **new** QueueingConsumer(channel);  
34.         channel.basicConsume(queueName, **true**, consumer);  
35.   
36.         **while** (**true**)  
37.         \{  
38.             QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
39.             String message = **new** String(delivery.getBody());  
40.   
41.             System.out.println(" \[x\] Received '" + message + "'");  
42.         \}  
43.     \}  
44.   
45.     /\*\* 
46.      \* 随机产生一种日志类型 
47.      \*  
48.      \* @return 
49.      \*/  
50.     **private** **static** String getSeverity()  
51.     \{  
52.         Random random = **new** Random();  
53.         **int** ranVal = random.nextInt(3);  
54.         **return** SEVERITIES\[ranVal\];  
55.     \}  
56. \}

接收端随机设置一个日志严重级别（binding\_key）。。。

我开启了3个接收端程序，两个准备接收error类型日志，一个接收info类型日志，然后运行发送端程序

运行结果：

\[x\] Sent 'error\_log :d142b096-46c0-4380-a1d2-d8b2ac136a9c'  
 \[x\] Sent 'error\_log :55ee1fc4-c87c-4e5e-81ba-49433890b9ce'  
 \[x\] Sent 'error\_log :d01877d6-87c7-4e0a-a109-697d122bc4c9'  
 \[x\] Sent 'error\_log :b42471b1-875c-43f1-b1ea-0dd5b49863f3'  
 \[x\] Sent 'info\_log :a6c1bc87-efb0-43eb-8314-8a74c345ed05'  
 \[x\] Sent 'info\_log :b6a84b6a-353e-4e88-8c23-c791d93b44be'

\------------------------------------------------------------------------------------

\[\*\] Waiting for error logs. To exit press CTRL+C  
 \[x\] Received 'error\_log :d142b096-46c0-4380-a1d2-d8b2ac136a9c'  
 \[x\] Received 'error\_log :55ee1fc4-c87c-4e5e-81ba-49433890b9ce'  
 \[x\] Received 'error\_log :d01877d6-87c7-4e0a-a109-697d122bc4c9'  
 \[x\] Received 'error\_log :b42471b1-875c-43f1-b1ea-0dd5b49863f3'

\------------------------------------------------------------------------------------

\[\*\] Waiting for info logs. To exit press CTRL+C  
 \[x\] Received 'info\_log :a6c1bc87-efb0-43eb-8314-8a74c345ed05'  
 \[x\] Received 'info\_log :b6a84b6a-353e-4e88-8c23-c791d93b44be'

可以看到我们实现了博文开头所描述的特性，接收者可以自定义自己感兴趣类型的日志。

其实文章这么长就在说：发送消息时可以设置routing\_key，接收队列与转发器间可以设置binding\_key，接收者接收与binding\_key与routing\_key相同的消息。

[RabbitMQ _]: http://blog.csdn.net/lmj623565791/article/details/37657225
[Center]: /images/20220721/2bed41dc59104d5ba51aefad36d43012.png
[Center 1]: /images/20220721/3e0bf6c523b24259ac50973b47e360e2.png
[Center 2]: /images/20220721/51deae676a4347968e25868b91bfaf23.png
[view plain]: http://blog.csdn.net/lmj623565791/article/details/37669573#