多个消费者服务未并行使用RabbitMQ队列 如何提高吞吐量

编程问答 2022-06-04

问题描述

我正在开发NodeJS和RabbitMQ，其中-

  • 主NodeJS服务（一个实例）将数据推送到队列中。
  • 从属NodeJS服务（多个实例）使用数据并对其进行处理。
  • 正在使用默认交换。

从属服务在PM2群集模式下运行，这意味着我有8个实例用于从属服务的运行。

我的期望是，随着主服务开始通过队列推送数据，从服务应开始异步使用它们。

例如，如果主服务正在队列中推送10个作业，并且每个作业需要5秒钟才能完成，则从属服务器需要50秒钟才能完成作业。

这完全违反了使用多个奴隶的目的，因为我绝对希望奴隶一次承办8个工作。

根据RabbitMQ仪表板，以上设置将创建-

  • 9个连接（1个主+ 8个从）
  • 9个通道（1个主+ 8个从）
  • 1个队列

整个设置使用默认交换。

我的问题是-

为什么从站无法从队列中异步读取数据？

即使我将noAck设置为true，在处理当前项目之前，队列中的下一个项目也不会被拾取

我的意图是通过用户多个从属实例来扩大队列使用率，但我认为我在这里遗漏了一些东西。

这是代码库-

const rabbitMq = require("amqplib");

class RabbitMQClient {

    async connect() {
        this.connection = await rabbitMq.connect("amqp://admin:password@localhost");
        this.channel = await this.connection.createChannel();
        await this.channel.assertQueue("TEST_QUEUE");
    }
}


// MASTER CODE (This Runs In Fork Mode - 1 Instance)
const master_client = new RabbitMQClient();
master_client.connect().then(() => {
    // sending in 10 messages
    for (let index = 1; index <= 10; index++) {
        const data = Buffer.from(index.toString(),"utf8");
        master_client.channel.sendToQueue("TEST_QUEUE",data);
    }
});


// SLAVE CODE (This Runs In Cluster Mode - 8 Instances)
const slave_client = new RabbitMQClient();
// connect to rabbitmq
slave_client.connect().then(() => {
    // consume the messages
    slave_client.channel.consume("TEST_QUEUE",(data) => {
        // timeout to add delay
        setTimeout(() => {
            RabbitMQClient._channel.ack(data);
        },5000);
    });
});

从属输出-

33|slave | 2020-11-02 13:19:09.293 +00:00: recieved message - 1  (13-19-09)
34|slave | 2020-11-02 13:19:14.293 +00:00: recieved message - 2  (13-19-14)
35|slave | 2020-11-02 13:19:19.299 +00:00: recieved message - 3  (13-19-19)
36|slave | 2020-11-02 13:19:24.299 +00:00: recieved message - 4  (13-19-24)
37|slave | 2020-11-02 13:19:29.300 +00:00: recieved message - 5  (13-19-29)
38|slave | 2020-11-02 13:19:34.299 +00:00: recieved message - 6  (13-19-34)
39|slave | 2020-11-02 13:19:39.301 +00:00: recieved message - 7  (13-19-39)
40|slave | 2020-11-02 13:19:44.301 +00:00: recieved message - 8  (13-19-44)
33|slave | 2020-11-02 13:19:49.299 +00:00: recieved message - 9  (13-19-49)
34|slave | 2020-11-02 13:19:54.300 +00:00: recieved message - 10 (13-19-54)

如果您注意到，则不同的从属以循环方式接收消息，但它们正在同步工作。

谢谢！

解决方法

RabbitMQ中的队列是单线程的，无法分派消息 同时并行。但这并不意味着消费者不会同时进行处理。当一个使用者使用时，绑定到同一队列的其他使用者可以同时接收消息并开始使用。

如何提高吞吐量

通常，以下提示总是有帮助的：

  • 将队列划分到不同的内核上。您可以使用RabbitMQ Sharding插件，该插件可以自动将队列分片，也可以手动完成
  • 为每个使用者设置最佳的预取计数，这可以降低使用者获取消息的网络成本
    • 每个消费者
  • 1个渠道。与队列相同，通道也是单线程的

有关更多详细信息，您可以访问here

node-amqplib