RabbitMQ 主题模式

待我称王封你为后i 2024-03-25 20:17 81阅读 0赞

##### 一、主题模式 #####

官方内容参考：[http://www.rabbitmq.com/tutorials/tutorial-five-java.html][http_www.rabbitmq.com_tutorials_tutorial-five-java.html]

跟路由模式类似，只不过路由模式是指定固定的路由键，而主题模式是可以模糊匹配路由键，类似于SQL中=和like的关系。

![36fb1c089df64540a170a08b328b827c.png][]

##### 二、topic交换机 #####

这个就不解释了，配置路由键的时候可以配置 \*, \# 来模糊匹配

* (star) can substitute for exactly one word.
    
    * 号表示可以精确匹配一个单词
    
    # (hash) can substitute for zero or more words.
    
    # 号可以匹配0个或者多个单词

举例：如上图的主题模式中， Q1绑定*.orange.*路由键，Q2绑定两个路由键，分别是*.*.rabbit以及lazy.\#

如果生产者发送路由键为quick.orange.rabbit消息，C1和C2都可以接收到。

如果为lazy.orange.elephant， C1和C2都可以接收到

如果为quick.orange.fox， 只要C1可以接收到

如果为lazy.brown.fox， 只有C2可以接收到

如果为lazy.pink.rabbit， C1，C2都可以

如果为quick.brown.fox， 都不会接收到

##### 三、代码演示 #####

只是路由键可以模糊匹配了，先设定为生产者发送quick.orange.rabbit，消费者1绑定到*.orange.*，消费者2绑定到*.*.rabbit以及lazy.\#

###### 生产者 ######

package com.rabbitmq.topic;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.util.ConnectionUtil;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * 生产者
     */
    public class Send {
    
        private static final String EXCHANGE_NAME = "test_exchange_topic";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 获取连接
            Connection connection = ConnectionUtil.getConnection();
            // 从连接开一个通道
            Channel channel = connection.createChannel();
            // 声明一个topic路由交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
    
            // 发送消息
            String message = "hello, quick.orange.rabbit";
            channel.basicPublish(EXCHANGE_NAME, "quick.orange.rabbit", null, message.getBytes());
            System.out.println(" [x] Sent message : '" + message + "'");
    
            channel.close();
            connection.close();
    
        }
    }

###### 消费者1 ######

package com.rabbitmq.topic;
    
    import com.rabbitmq.client.*;
    import com.rabbitmq.util.ConnectionUtil;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * 消费者
     */
    public class Recv1 {
    
        private static final String QUEUE_NAME = "test_queue_topic_1";
        private static final String EXCHANGE_NAME = "test_exchange_topic";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 获取连接
            Connection connection = ConnectionUtil.getConnection();
    
            // 打开通道
            Channel channel = connection.createChannel();
    
            // 申明要消费的队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.orange.*");
    
            // 这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。换句话说，在接收到该Consumer的ack前，他它不会将新的Message分发给它。
            channel.basicQos(1);
    
            // 创建一个回调的消费者处理类
            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);
                    System.out.println(" [1] Received '" + message + "'");
    
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        System.out.println(" [1] done ");
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
    
            // 消费消息
            channel.basicConsume(QUEUE_NAME, false, consumer);
    
        }
    
    }

###### 消费者2 ######

package com.rabbitmq.topic;
    
    import com.rabbitmq.client.*;
    import com.rabbitmq.util.ConnectionUtil;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * 消费者2
     */
    public class Recv2 {
    
        private static final String QUEUE_NAME = "test_queue_topic_2";
        private static final String EXCHANGE_NAME = "test_exchange_topic";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 获取连接
            Connection connection = ConnectionUtil.getConnection();
    
            // 打开通道
            Channel channel = connection.createChannel();
    
            // 申明要消费的队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定队列到交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.*.rabbit");
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "lazy.#");
    
            // 这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。换句话说，在接收到该Consumer的ack前，他它不会将新的Message分发给它。
            channel.basicQos(1);
    
            // 创建一个回调的消费者处理类
            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);
                    System.out.println(" [2] Received '" + message + "'");
    
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        System.out.println(" [2] done ");
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
    
            // 消费消息
            channel.basicConsume(QUEUE_NAME, false, consumer);
        }
    
    }

##### 四、测试结果 #####

提前在管理控制台创建一个topic交换机，或者先执行一次生产者（执行时会判断交换机是否存在，不存在则创建交换机），这样保证交换机存在，不然直接启动消费者会提示交换机不存在。

![c39f7d73269345a9b67ac4bc653111e9.png][]

###### 注意点 ######

如果在没有队列绑定在交换机上面时，往交换机发送消息会丢失，之后绑定在交换机上面的队列接
    收不到之前的消息，也就是先执行第一次发送，创建了交换机，但是还没有队列绑定在交换机上面，
    如果这次发送的消息就会丢失。

然后启动两个消费者，再执行生产者。

Send

[x] Sent message : 'hello, quick.orange.rabbit'

Recv

[1] Received 'hello, quick.orange.rabbit'
     [1] done

Recv2

[2] Received 'hello, quick.orange.rabbit'
     [2] done

我们可以看到生产者往quick.orange.rabbit路由键发送消息时，两个消费者都收到了消
    息，说明都匹配到了。

###### 其他情况测试结果 ######

1.  测试生产者发送lazy.orange.elephant

结果：两个消费者也都能收到消息。

1.  测试生产者发送quick.orange.fox

结果：只要消费者1能接收到。

1.  测试生产者发送lazy.brown.fox

结果：只有消费者2能接收到。

其他情况自行测试。

[http_www.rabbitmq.com_tutorials_tutorial-five-java.html]: http://www.rabbitmq.com/tutorials/tutorial-five-java.html
[36fb1c089df64540a170a08b328b827c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/e1db61857a544febb3b195823ca6f689.png
[c39f7d73269345a9b67ac4bc653111e9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/1bbd91d4125f4b41bc643de55213500f.png