RabbitMQ系列（五）Topic模型

Jensen_97

发布于 2023-07-20 14:53:17

1640

发布于 2023-07-20 14:53:17

文章被收录于专栏：技术客栈

Topic 模型

上一篇文章《RabbitMQ系列（四）通信模型之路由模型》中，简单的介绍了一下RabbitMQ的路由模型。这篇文章来学习一下RabbitMQ中的topic模型，Topic 模型是 RabbitMQ 的高级模型之一，Topic 模型使用了通配符的概念，可以匹配更灵活的路由规则。topic模式相当于是对路由模式的一个升级，topic模式主要就是在匹配的规则上可以实现模糊匹配。

在 Topic 模型中，生产者将消息发送到交换机，交换机根据消息的 routing key 将消息转发到对应的队列中。与 Direct 模型不同的是，Topic 模型中 routing key 支持通配符匹配，其中 '*' 可以匹配一个单词，'#' 可以匹配多个单词。例如，"order.*" 可以匹配 "order.create"，"order.delete" 等消息，而 "order.#" 可以匹配 "order.create.one"，"order.delete.two" 等消息。

适用场景

Topic 模型适用于需要灵活的消息路由规则的场景，例如：

新闻网站订阅分类消息；
电商网站订阅商品分类消息；
金融机构订阅股票市场消息等。

演示

生产者

// 生产者
public class Producer {
    private static final String EXCHANGE_NAME = "exchange_topic_1";
    private static final String EXCHANGE_ROUTING_KEY1 = "topic.km";
    private static final String EXCHANGE_ROUTING_KEY2 = "topic.km.001";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        for (int i = 0; i < 100; i++) {
            // topic在路由模型的基础上，只有路由的key发生改变，其余的都不变
            if (i % 2 == 0) {
                channel.basicPublish(EXCHANGE_NAME, EXCHANGE_ROUTING_KEY1, MessageProperties.PERSISTENT_TEXT_PLAIN, ("topic模型发送的第 " + i + " 条信息").getBytes());
            } else {
                channel.basicPublish(EXCHANGE_NAME, EXCHANGE_ROUTING_KEY2, MessageProperties.PERSISTENT_TEXT_PLAIN, ("topic模型发送的第 " + i + " 条信息").getBytes());
            }
        }
        channel.close();
        connection.close();
    }
}

消费者

// 消费者1
public class Consumer1 {
    private static final String QUEUE_NAME = "queue_topic_1";
    private static final String EXCHANGE_NAME = "exchange_topic_1";
    private static final String EXCHANGE_ROUTING_KEY = "topic.*";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, EXCHANGE_ROUTING_KEY);
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1接收到的消息是：" + new String(body));
            }
        };
        channel.basicConsume(QUEUE_NAME, true, defaultConsumer);
    }
}

// 消费者2
public class Consumer2 {
    private static final String QUEUE_NAME = "queue_topic_2";
    private static final String EXCHANGE_NAME = "exchange_topic_1";
    private static final String EXCHANGE_ROUTING_KEY = "topic.#";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, EXCHANGE_ROUTING_KEY);
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2接收到的消息是：" + new String(body));
            }
        };
        channel.basicConsume(QUEUE_NAME, true, defaultConsumer);
    }
}