Java初识RabbitMQ一交换机（topic exchange）

古城微笑少年丶 2023-07-04 12:59 3阅读 0赞

推荐：[消息中间件汇总][Link 1]

## Java初识RabbitMQ一交换机（topic exchange） ##

首先看看`AMQP`协议，对`RabbitMQ`的架构会更了解。

[深入理解AMQP协议][AMQP]

创建一个`Maven`项目，根据自己服务器RabbitMQ的版本导入相应的包。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70]

<dependency>
    			<groupId>com.rabbitmq</groupId>
    			<artifactId>amqp-client</artifactId>
    			<version>3.6.5</version>
    		</dependency>

**主题交换机**

前面提到的 direct exchange的路由规则是严格意义上的匹配，换言之 Routing Key 必须与 Binding Key 相匹配的时候才将消息传送给 Queue。

而topic exchange的路由规则是一种模糊匹配，可以通过通配符满足一部分规则就可以传送。

它的约定是：

1）binding key 中可以存在两种特殊字符 `“*”` 与`“#”`，用于做模糊匹配，其中 `“*”` 用于匹配一个单词，`“#”`用于匹配多个单词（可以是零个）。

2）routing key 为一个句点号 “.” 分隔的字符串（我们将被句号 “. ” 分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”，binding key 与 routing key 一样也是句点号 “.” 分隔的字符串。

![在这里插入图片描述][70]  
当生产者发送消息 Routing Key=F.C.E 的时候，这时候只满足 Queue1，所以会被路由到 Queue1 中，如果 Routing Key=A.C.E 这时候会被同时路由到 Queue1 和 Queue2 中，如果 Routing Key=A.F.B 时，这里只会发送一条消息到 Queue2 中。

主题交换机拥有非常广泛的用户案例。无论何时，当一个问题涉及到那些想要有针对性的选择需要接收消息的 多消费者 / 多应用（multiple consumers/applications） 的时候，主题交换机都可以被列入考虑范围。

**生产端**

package com.kaven.rabbitmq.exchange.topicExchange;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class TopicProducer { 
    
        // 自己服务器的IP
        private static String ip = "IP";
        // RabbitMQ启动的默认端口，也是应用程序进行连接RabbitMQ的端口
        private static int port = 5672;
        // RabbitMQ有一个 "/" 的虚拟主机
        private static String virtualHost = "/";
    
        // topic exchange ，RabbitMQ提供的topic exchange
        private static String exchangeName = "amq.topic";
        // exchange type
        private static String exchangeType= "topic";
    
        // 交换机路由的routingKey
        private static String[] routingKey = { "test.kaven.wyy" , "test" , "test.topic"};
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            // 1 创建一个连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost(ip);
            connectionFactory.setPort(port);
            connectionFactory.setVirtualHost(virtualHost);
    
            // 2 创建连接
            Connection connection = connectionFactory.newConnection();
    
            // 3 创建Channel
            Channel channel = connection.createChannel();
    
            // 4 发送消息
            for (int i = 0; i < routingKey.length; i++) { 
                String msg = "RabbitMQ：Topic Exchange 发送数据 , routingKey："+routingKey[i];
                channel.basicPublish(exchangeName ,routingKey[i] ,null, msg.getBytes());
            }
    
            // 5 关闭连接
            channel.close();
            connection.close();
        }
    }

**消费端**

这里创建两个消费者，为每个消费者创建一个线程，以便监听生产端发送过来的消息。

package com.kaven.rabbitmq.exchange.topicExchange;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.QueueingConsumer;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class TopicConsumer { 
    
        // 自己服务器的IP
        private static String ip = "IP";
        // RabbitMQ启动的默认端口，也是应用程序进行连接RabbitMQ的端口
        private static int port = 5672;
        // RabbitMQ有一个 "/" 的虚拟主机
        private static String virtualHost = "/";
    
        // topic exchange ，RabbitMQ提供的topic exchange
        private static String exchangeName = "amq.topic";
        // exchange type
        private static String exchangeType= "topic";
        // 队列名
        private static String[] queueName = { "queue_#" , "queue_*"};
        // 队列与交换机绑定的routingKey
        private static String[] routingKey = { "test.#" , "test.*"};
    
        public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { 
            // 1 创建一个连接工厂
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost(ip);
            connectionFactory.setPort(port);
            connectionFactory.setVirtualHost(virtualHost);
    
            // 2 创建连接
            Connection connection = connectionFactory.newConnection();
    
            // 3 创建Channel
            Channel channel = connection.createChannel();
    
            for (int i = 0; i < queueName.length; i++) { 
                // 4 定义Queue ，将Queue绑定到direct exchange
                channel.queueDeclare(queueName[i],true , false , false , null);
                channel.queueBind(queueName[i] , exchangeName , routingKey[i]);
            }
    
    
            // 5 创建消费者
            QueueingConsumer consumer0 = new QueueingConsumer(channel);
            QueueingConsumer consumer1 = new QueueingConsumer(channel);
    
            // 6 设置
            channel.basicConsume(queueName[0] , true , consumer0);
            channel.basicConsume(queueName[1] , true , consumer1);
    
            // 7 接收消息
            Thread thread0 = new Thread(new MyRunnable(consumer0 , routingKey[0]));
            Thread thread1 = new Thread(new MyRunnable(consumer1 , routingKey[1]));
    
            thread0.start();
            thread1.start();
    
            thread0.join();
            thread1.join();
        }
    }

package com.kaven.rabbitmq.exchange.topicExchange;
    
    import com.rabbitmq.client.QueueingConsumer;
    
    public class MyRunnable implements Runnable { 
    
        private QueueingConsumer consumer;
        private String routingKey;
    
        public MyRunnable(QueueingConsumer consumer , String routingKey) { 
            this.consumer = consumer;
            this.routingKey = routingKey;
        }
    
        @Override
        public void run() { 
            while(true){ 
                try { 
                    QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                    String msg = new String(delivery.getBody());
                    System.out.println(routingKey+"[收到]-"+msg);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    }

**测试**

因为这里使用的是`RabbitMQ`提供给我们的topic exchange，所以我们无需自己定义。  
因为交换机已经定义好了，所以无论先启动生产端还是消费端，消费端都可以成功收到消息。  
消费端输出如下：

test.#[收到]-RabbitMQ：Topic Exchange 发送数据 , routingKey：test.kaven.wyy
    test.*[收到]-RabbitMQ：Topic Exchange 发送数据 , routingKey：test.topic
    test.#[收到]-RabbitMQ：Topic Exchange 发送数据 , routingKey：test
    test.#[收到]-RabbitMQ：Topic Exchange 发送数据 , routingKey：test.topic

很明显输出结果符合上面对topic exchange路由规则的解释。

看看`RabbitMQ Management`。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 3]

[Link 1]: https://kaven.blog.csdn.net/article/details/104223534
[AMQP]: https://blog.csdn.net/qq_37960603/article/details/104225465
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2020020922542160.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz,size_16,color_FFFFFF,t_70
[70]: https://img-blog.csdn.net/20181022131252616?watermark/2/text/aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNzY0MTgzMg==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200211161708623.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200211162001755.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200211162017738.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3OTYwNjAz,size_16,color_FFFFFF,t_70