RabbitMQ06_Topic模型

「爱情、让人受尽委屈。」 2022-12-19 11:37 98阅读 0赞

**RabbitMQ06\_Topic模型**

上一个模型（订阅直连（Direct）模型）实现了同的消息可以被不同队列消费的需求，但是它还是不够灵活。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JMVV8xMTE_size_16_color_FFFFFF_t_70_pic_left]  
**Topic 模型允许队列绑定 Routing key 时使用通配符，这种模型的 Routingkey 一般由一个或多个单词组成，多个单词以"."分隔，例如 item.insert**

**可以使用 \* 号匹配一个单词，使用 \# 号匹配多个单词**

**代码示例：**

*  **消息生产者：**

public static void main(String[] args) throws IOException { 
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare("topics", "topic");
        String routingkey = "user.save.all";
        channel.basicPublish("topics", routingkey, null, ("topic 动态路由模型,route key:["+routingkey+"]").getBytes());
        RabbitMQUtils.closeConnectionAndChannel(channel, connection);
    }

*  **消息消费者1：**

public static void main(String[] args) throws IOException { 
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare("topics", "topic");
        String queue = channel.queueDeclare().getQueue();
        channel.queueBind(queue, "topics", "user.*");
        channel.basicConsume(queue, true, new DefaultConsumer(channel){ 
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                throws IOException { 
                System.out.println("消费者1："+new String(body));
            }
        });
    }

*  **消息消费者2：**

public static void main(String[] args) throws IOException { 
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare("topics", "topic");
        String queue = channel.queueDeclare().getQueue();
        channel.queueBind(queue, "topics", "user.#");
        channel.basicConsume(queue, true, new DefaultConsumer(channel){ 
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                throws IOException { 
                System.out.println("消费者2："+new String(body));
            }
        });
    }

*  **测试分别发送 Routing key 基于 user.save 和 user.save.aaa 的两条消息，结果：**

消费者1只接收到1条消息，而消费者2接收到了两条消息：

消费者1：topic 动态路由模型,route key:[user.save]

消费者2：topic 动态路由模型,route key:[user.save]
    消费者2：topic 动态路由模型,route key:[user.save.aaa]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0JMVV8xMTE_size_16_color_FFFFFF_t_70_pic_left]: /images/20221120/8d0c448811424d74aa911999968ea456.png