RabbitMQ（五）：Exchange交换器--topic

Love The Way You Lie 2022-04-24 20:24 185阅读 0赞

内容翻译自：[RabbitMQ Tutorials Java版][RabbitMQ Tutorials Java]

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

[RabbitMQ（一）：Hello World程序][RabbitMQ_Hello World]

[RabbitMQ（二）：Work Queues、循环分发、消息确认、持久化、公平分发][RabbitMQ_Work Queues]

[RabbitMQ（三）：Exchange交换器--fanout][RabbitMQ_Exchange_--fanout]

[RabbitMQ（四）：Exchange交换器--direct][RabbitMQ_Exchange_--direct]

[RabbitMQ（五）：Exchange交换器--topic][RabbitMQ_Exchange_--topic]

[RabbitMQ（六）：回调队列callback queue、关联标识correlation id、实现简单的RPC系统][RabbitMQ_callback queue_correlation id_RPC]

[RabbitMQ（七）：常用方法说明 与 学习小结][RabbitMQ_ _]

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

### Topics： ###

在上一篇博客中我们改进了我们的日志系统：使用`direct`路由器替代了`fanout`路由器，从而可以选择性地接收日志。

尽管使用direct路由器给我们的日志系统带来了改进，但仍然有一些限制：不能基于多种标准进行路由。

在我们的日志系统中，我们可能不仅需要根据日志的严重级别来接收日志，而且有时想基于日志来源进行路由。如果你知道[syslog][]这个Unix工具，你可能了解这个概念，`sysylog`会基于日志严重级别（`info/warn/crit...`）和设备（`auth/cron/kern...`）进行日志分发。

如果我们可以监听来自`corn`的错误日志，同时也监听`kern`的所有日志，那么我们的日志系统就会更加灵活。

为了实现这个功能，我们需要了解一个复杂的路由器：`topic`路由器。

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

### 主题路由器（Topic Exchange）： ###

发送到`topic`路由器的消息的路由键`routing_key`不能任意给定：它必须是一些单词的集合，中间用点号`.`分割。这些单词可以是任意的，但通常会体现出消息的特征。一些有效的路由键示例：`stock.usd.nyse`，`nyse.vmw`，`quick.orange.rabbit`。这些路由键可以包含很多单词，但路由键总长度不能超过255个字节。

绑定键`binding key`也必须是这种形式。`topic`路由器背后的逻辑与`direct`路由器类似：以特定路由键发送的消息将会发送到所有绑定键与之匹配的队列中。但绑定键有两种特殊的情况：

（1）\*（星号）仅代表一个单词

（2）\#（井号）代表任意个单词

下图可以很好地解释这两个符号的含义：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E3NDUyMzM3MDA_size_16_color_FFFFFF_t_70][]

对于上图的例子，我们将会发送描述动物的消息。这些消息将会以由三个单词组成的路由键发送。路由键中的第一个单词描述了速度，第二个描述了颜色，第三个描述了物种：`<speed>.<colour>.<species>`。

我们创建了三个绑定，Q1的绑定键为`*.orange.*`，Q2的绑定键有两个，分别是`*.*.rabbit`和`lazy.#`。

上述绑定关系可以描述为：

（1）Q1关注所有颜色为`orange`的动物。

（2）Q2关注所有的`rabbit`，以及所有的`lazy`的动物。

如果一个消息的路由键是`quick.orange.rabbit`，那么Q1和Q2都可以接收到，路由键是`lazy.orange.elephant`的消息同样如此。但是，路由键是`quick.orange.fox`的消息只会到达Q1，路由键是`lazy.brown.fox`的消息只会到达Q2。注意，路由键为`lazy.pink.rabbit`的消息只会到达Q2一次，尽管它匹配了两个绑定键。路由键为`quick.brown.fox`的消息因为不和任意的绑定键匹配，所以将会被丢弃。

假如我们不按常理出牌：发送一个路由键只有一个单词或者四个单词的消息，像`orange`或者`quick.orange.male.rabbit`，这样的话，这些消息因为不和任意绑定键匹配，都将会丢弃。但是，`lazy.orange.male.rabbit`消息因为和`lazy.#`匹配，所以会到达Q2，尽管它包含四个单词。

> **Topic exchange：:**
> 
> `Topic exchange`非常强大，可以实现其他任意路由器的功能。
> 
> 当一个队列以绑定键`#`绑定，它将会接收到所有的消息，而无视路由键（实际是绑定键`#`匹配了任意的路由键）。----这和`fanout`路由器一样了。
> 
> 当`*`和`#`这两个特殊的字符不出现在绑定键中，`Topic exchange`就会和`direct exchange`类似了。

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

### 放在一块： ###

我们将会在我们的日志系统中使用主题路由器`Topic exchange`，并假设所有的日志消息以两个单词`<facility>.<severity>`为路由键。

代码和上个教程几乎一样。

生产者`EmitLogTopic.java`：

import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.Channel;
    
    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, BuiltinExchangeType.TOPIC);
    
                //定义路由键和消息
                String routingKey = "";
                String message = "msg.....";
    
                //发布消息
                channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
                System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");
    
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (connection != null) {
                    try {
                        connection.close();
                    } catch (Exception ignore) {
                    }
                }
            }
        }
    }

消费者`ReceiveLogsTopic.java`：

import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    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, BuiltinExchangeType.TOPIC);
            String queueName = channel.queueDeclare().getQueue();
    
            //
            String bingingKeys[] = {""};
    
            for (String bindingKey : bingingKeys) {
                channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
            }
    
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    
            //监听消息
            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 '" + envelope.getRoutingKey() + "':'" + message + "'");
                }
            };
            channel.basicConsume(queueName, true, consumer);
        }
    }

现在，可以动手实验了。  
开头提到的：日志严重级别`info/warn/crit...`和设备`auth/cron/kern...`。

消费者：  
将`String bingingKeys[] = {""}`改为`String bingingKeys[] = {"#"}`，启动第一个消费者；  
再改为`String bingingKeys[] = {"kern.*"}`，启动第二个消费者；  
再改为`String bingingKeys[] = {"*.critical"}`，启动第三个消费者；  
再改为`String bingingKeys[] = {"kern.*", "*.critical"}`，启动第四个消费者。

生产者，发送多个消息，如：  
路由键为`kern.critical` 的消息：`A critical kernel error`；  
路由键为`kern.info` 的消息：`A kernel info`；  
路由键为`kern.warn` 的消息：`A kernel warning`；  
路由键为`auth.critical` 的消息：`A critical auth error`；  
路由键为`cron.warn` 的消息：`A cron waning`；  
路由键为`cron.critical` 的消息：`A critical cron error`；

试试最后的结果：第一个消费者将会接收到所有的消息，第二个消费者将会`kern`的所有严重级别的日志，第三个消费者将会接收到所有设备的`critical`消息，第四个消费者将会接收到`kern`设备的所有消息和所有`critical`消息。

下篇博客中，我们将会学习如何让消息往返，以此来作为一个远程过程调用（RPC）。

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

### 说明 ###

①与原文略有出入，如有疑问，请参阅原文

②原文均是编译后通过javacp命令直接运行程序，我是在IDE中进行的，相应的操作做了修改。

[RabbitMQ Tutorials Java]: https://link.jianshu.com?t=http://www.rabbitmq.com/tutorials/tutorial-one-java.html
[RabbitMQ_Hello World]: https://blog.csdn.net/a745233700/article/details/89680776
[RabbitMQ_Work Queues]: https://blog.csdn.net/a745233700/article/details/89680957
[RabbitMQ_Exchange_--fanout]: https://blog.csdn.net/a745233700/article/details/89681388
[RabbitMQ_Exchange_--direct]: https://blog.csdn.net/a745233700/article/details/89681621
[RabbitMQ_Exchange_--topic]: https://blog.csdn.net/a745233700/article/details/89681801
[RabbitMQ_callback queue_correlation id_RPC]: https://blog.csdn.net/a745233700/article/details/89681941
[RabbitMQ_ _]: https://blog.csdn.net/a745233700/article/details/89703132
[syslog]: https://link.jianshu.com?t=http://en.wikipedia.org/wiki/Syslog
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E3NDUyMzM3MDA_size_16_color_FFFFFF_t_70]: /images/20220212/c16a4999e07b4840b9cb8956006ce3e8.png