RabbitMQ教程-Topic Exchange类型的基本使用demo

拼搏现实的明天。 2023-01-07 16:19 125阅读 0赞

echo编辑整理，欢迎转载，转载请声明文章来源。欢迎添加echo微信(微信号：t2421499075) 交流学习。

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

> 上两篇文章中，我们使用Direct Exchange和Fanout Exchange类型推送了消息，但是我们发现这两种类型在某些情况下并不适合，比如：我们MQ中有一百个不同队列订阅了我们的某个生产者，但是在某一次的时候，我们生产者只希望其中一部分对了能接受消息，那么我们的Fanout Exchange直接广播就会造成信息资源浪费，如果我们选择Direct Exchange的时候，那我们要发送给对应的几十个队列就会要不断设置，重复工作很多，也很容易出错。但是要是能够根据我们的队列RoutingKey来模糊匹配是不是很我们的问题就完美解决了啊。本文就是讲解这种模式

### 使用Topic Exchange根据规则推送一条消息 ###

*  准备工作  
    既然我们要根据规则推送给某一批队列，我们要先建立一批队列，指定路由器类型为Topic Exchange，然后我们使用规则定义routing key。我们这里准备了三个队列

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsZWNobw_size_16_color_FFFFFF_t_70]

并通过routing key进行绑定到topic类型的Exchange上

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsZWNobw_size_16_color_FFFFFF_t_70 1]

### routing key作用解析 ###

routing key的命名是我们Topic Exchange类型的最关键部分，在推送消息的时候，如果是使用topic类型的Exchange可以直接不指定队列，所以这个时候routing key就成为了我们推送消息的关键。为了消息能够更好的推送接受，在这个地方我们肯定要规范的定义routing kye的名称。在我们这里，我们看了总共有三个routing key

*  com.echo.\* :当推送消息时routing key为com.echo.任意字符的时候，队列queue\_topic1对应的队列一定能接受到消息
 *  com.echo.level2 :当推送消息时routing key为com.echo.level2，队列queue\_topic2对应的队列才会接受到消息
 *  `#` :当推送消息时routing key为任意值，队列queue\_topic2对应的队列都能接受到消息

其中 \* 匹配任意一个单词，\# 匹配零个或者多个单词。

代码案例

package com.example.demo;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * @author echo
     * @date 2021-01-14 14:35
     */
    public class TopicProductTest {
        
    
        private static final String EXCHANGE_NAME = "exchange_topic";
        private static final String ROUTING_KEY = "com.echo.level2";
        private static final String IP_ADDRESS = "192.168.230.131";
        private static final int PORT = 5672;
    
    
        public static void main(String[] args) throws IOException, TimeoutException {
        
    
            // 创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            // 设置RabbitMQ的链接参数
            factory.setHost(IP_ADDRESS);
            factory.setPort(PORT);
            factory.setUsername("echo");
            factory.setPassword("123456");
            // 和RabbitMQ建立一个链接
            Connection connection = factory.newConnection();
            // 创建一个频道
            Channel channel = connection.createChannel();
            // 创建一个 type="direct" 、持久化的、非自动删除的交换器
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC, true, false, null);
            // 发送一条持久化的消息: topic hello world !
            String message = "topic hello world !";
            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
            // 关闭资源
            channel.close();
            connection.close();
    
        }
    
    }

按照上面routing key的作用，我们这个实例推送的消息，对应的三个队列应该都会受到消息，我们运行了看看结果

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsZWNobw_size_16_color_FFFFFF_t_70 2]

> 最终结果和我们的作用域的描述符合

*  我们还可以使用消费者来验证一下  
    消费者代码

package com.example.demo;
    
    import com.rabbitmq.client.*;
    import lombok.SneakyThrows;
    
    import java.io.IOException;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.TimeoutException;
    
    /**
     * @author tang.sl
     * @date 2021-01-14 15:05
     */
    public class TopicConsumerTest {
        
    
        private static final String EXCHANGE_NAME = "exchange_topic";
        private static final String QUEUE_NAME = "queue_topic1";
        private static final String IP_ADDRESS = "192.168.230.131";
        private static final int PORT = 5672;
    
        public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        
    
            ConnectionFactory factory = new ConnectionFactory();
            // 设置RabbitMQ的链接参数
            factory.setUsername("echo");
            factory.setPassword("123456");
            factory.setPort(PORT);
            factory.setHost(IP_ADDRESS);
    
            // 和RabbitMQ建立一个链接
            Connection connection = factory.newConnection();
            final Channel channel = connection.createChannel();
            //声明交换机 Fanout模式
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC, true, false, null);
            //进行绑定，指定消费那个队列
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "", null);
            Consumer consumer = new DefaultConsumer(channel) {
        
                @SneakyThrows
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
        
                    System.out.println("recv message: " + new String(body));
                    try {
        
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
            channel.basicConsume(QUEUE_NAME, consumer);
            //等待回调函数执行完毕之后 关闭资源
            TimeUnit.SECONDS.sleep(5);
            channel.close();
            connection.close();
        }
    }

运行之后结果一致

![在这里插入图片描述][20210118134643331.png]

### 总结 ###

topic能够对指定的一系列或者说一堆的队列发送消息，关键就是靠routing key的\*和\#这两个通配符的匹配作用

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsZWNobw_size_16_color_FFFFFF_t_70]: /images/20221119/4285c42d3287454f8b1b97693721a9d1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsZWNobw_size_16_color_FFFFFF_t_70 1]: /images/20221119/d0e511946b254f29b34b7a1e75d153f9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsZWNobw_size_16_color_FFFFFF_t_70 2]: /images/20221119/9e139c381c114a5caa6875a882021bd2.png
[20210118134643331.png]: /images/20221119/ff6bbf7ba66e4f34820f5495951dcc0d.png