Rabbitmqtopic 主题详解

发布时间：2023-05-19

一、基础概念

RabbitMQ 是一种使用 AMQP（Advanced Message Queuing Protocol）协议来实现消息队列（Message Queue）服务的中间件（Middleware）系统。它支持多种消息队列模型，其中之一是 Topic。Topic 是一种支持主题（Topic）概念的消息队列模型。在 Topic 模型中，消息可以被发送到一个这样的主题（Topic）中，所有订阅该主题的消费者都将收到该消息。而 RabbitMQ 通过 Exchange（交换机）来支持 Topic 模型。根据 Exchange 的不同绑定模式，可以实现以不同的主题来发送消息。 假设存在一个主题 topic.a.b，那么所有使用该主题的订阅者（Subscriber）都可以收到发送到该主题的消息。而 RabbitMQ 中，主题都是由一个或多个单词组成的路由键（Routing Keys）来表示的。所以，一个主题可以表示为用“.”分隔的多个单词组成的路由键，例如 topic.a.b.c 表示三个单词组成的主题。同时，在订阅主题时，也可以使用通配符（Wildcards）来表示一些特定的主题，例如 topic.# 表示一组以 topic 开头的主题。

二、Topic 模型的使用场景

在 RabbitMQ 中，使用 Topic 模型可以解决以下问题： 1、任务分发：一个任务被分发到多个消费者进行处理。例如，需要将爬取到的数据分发给多个处理模块进行处理。 2、关注事件：多个消费者可以订阅关注相同事件的主题。例如，在电商网站上，不同的用户可能对同一件商品感兴趣，而我们需要及时地通知他们在该商品上的优惠信息。 3、日志分析：将不同的日志事件路由到不同的消息队列中进行处理。例如，将用户行为日志、服务器日志和业务逻辑日志分别存储到不同的消息队列中。

三、Topic 模型的代码实现

1、发送消息

const amqp = require('amqplib/callback_api');
// 创建连接
amqp.connect('amqp://localhost', function (err, connection) {
  // 创建频道
  connection.createChannel(function (err, channel) {
    // 设置交换机类型为 topic
    let ex = 'topic_logs';
    let msg = process.argv.slice(2).join(' ') || 'Hello World!';
    let severity = process.argv.slice(2).join(' ') || 'anonymous.info';
    // 发布消息
    channel.assertExchange(ex, 'topic', { durable: false });
    channel.publish(ex, severity, Buffer.from(msg));
    console.log(" [x] Sent %s: '%s'", severity, msg);
    // 关闭频道和连接
    setTimeout(function () { 
      channel.close(); 
      connection.close(); 
    }, 500);
  });
});

2、接收消息

const amqp = require('amqplib/callback_api');
// 创建连接
amqp.connect('amqp://localhost', function (err, connection) {
  // 创建频道
  connection.createChannel(function (err, channel) {
    // 设置交换机类型为 topic
    let ex = 'topic_logs';
    let args = process.argv.slice(2);
    // 声明交换机和队列
    channel.assertExchange(ex, 'topic', { durable: false });
    channel.assertQueue('', { exclusive: true }, function (err, q) {
      // 根据传入参数绑定相应的路由键
      args.forEach(function (key) {
        channel.bindQueue(q.queue, ex, key);
      });
      // 监听消息队列
      channel.consume(q.queue, function (msg) {
        console.log(" [x] %s:'%s'", msg.fields.routingKey, msg.content.toString());
      }, { noAck: true });
    });
  });
});

四、Topic 模型的优化

在 RabbitMQ 中，实现 Topic 模型的最佳实践是合理地设计交换机和 Routing Keys。对于同一类主题的消息，应该尽可能地使用相同的 Routing Keys 策略。如果存在多个 Routing Keys，那么建议将其扇出（Fanout）为多个队列，避免重复消费等问题。 同时，在实现 Topic 模型时，应尽可能将消费者和生产者进行解耦。例如，可以使用 RPC（Remote Procedure Call，远程过程调用）方式实现消费者和生产者之间的通信，从而极大地提高消息队列的应用效率。

推荐文章