Nodejs RabbitMQ 服务器上的高性能

Question

我正在构建一个同时拥有一百万在线用户的分析系统。我使用消息代理等 RabbitMQ 来减少服务器容量

这是我的图表

我的系统包括 3 个组件。

发布服务器：（生产者） 这个系统是建立在 nodejs 上的。本系统的目的是将消息发布到queue

RabbitMQ 队列：该系统存储了publisher server 发送到的消息。之后，打开一个连接以从队列中为subscriber server 发送消息。

订阅服务器（消费者）：本系统接收来自queue的消息

发布服务器源代码

var amqp = require('amqplib/callback_api');
amqp.connect("amqp://localhost", function(error, connect) {
    if (error) {
        return callback(-1, null);
    } else {
       connect.createChannel(function(error, channel) {
       if (error) {
           return callback(-3, null);
       } else {
         var q = 'logs';
         var msg = data; // object
         // convert msg object to buffer 
         var new_msg = Buffer.from(JSON.stringify(msg), 'binary');

        channel.assertExchange(q, 'fanout', { durable: false });
        channel.publish(q, 'message_queues', new Buffer(new_msg));
       console.log(" [x] Sent %s", new_msg);
        return callback(null, msg);
      }
    });
   }
}); 

创建独家交换"message_queues" 与"fanout" 发送 向所有消费者广播

订阅服务器源代码

var amqp = require('amqplib/callback_api');
amqp.connect("amqp://localhost", function(error, connect) {
    if (error) {
        console.log('111');
    } else {
        connect.createChannel(function(error, channel) {
            if (error) {
                console.log('1');
            } else {
                var ex = 'logs';

                channel.assertExchange(ex, 'fanout', { durable: false });
                channel.assertQueue('message_queues', { exclusive: true }, function(err, q) {
                    if (err) {
                        console.log('123');
                    } else {
                        console.log(" [*] Waiting for messages in %s. To exit press CTRL+C", q.queue);
                        channel.bindQueue(q.queue, ex, 'message_queues');

                        channel.consume(q.queue, function(msg) {
                            console.log(" [x] %s", msg.content.toString());

                        }, { noAck: true });

                    }
                });
            }
        });
    }

});

接收来自"message_queues"exchange 的消息

当我实现发送消息时。系统运行良好，但是我尝试了该系统的基准测试性能（每秒大约 1000 个用户发送请求），然后系统出现了一些问题。系统似乎过载/缓冲区溢出（或某些事情无法正常工作）。

我 2 天前才读到有关 rabbitmq 的信息。我知道它的教程是基本示例，所以我需要帮助在现实世界中构建系统而不是......任何 解决方案和建议

希望我的问题有意义

Answer 1

您的问题很笼统。可能您应该提供更多详细信息以帮助识别瓶颈并帮助您。 所以，首先我认为你应该检查rabbit mq - 它是否是瓶颈。 有很多事情可能会出错：

1. 可以消费消息的消费者数量太少（我假设你使用消费者池）

2. 网络太慢

3. Rabbit MQ 的太多节点和 go do 磁盘之间复制了队列和消息（可以这样使用 rabbit mq）

4. 消费者无法真正处理消息，并且不断重新排队

所以，一般来说，在您的测试期间，您应该检查 rabbit mq 并看看那里发生了什么。

消息一旦进入队列就处于就绪状态，一旦发生这种情况，它将一直存在，直到连接到队列的消费者之一不会尝试获取消息进行处理

当其中一个消费者（兔子在他们之间进行循环）选择要处理的消息时，它的状态将变为 Unacknowledged 如果消费者未能处理消息，它将被兔子重新排队，以便另一个消费者有机会处理该消息。

当然，如果消费者成功处理了消息，消息就会从rabbit mq服务器上消失。

假设您已经安装了 rabbit mq web ui（我强烈推荐它，尤其是对于初学者） - 您可以直观地看到队列中发生的情况 - 您将看到有多少消息处于就绪状态，有多少消息未确认。 这将有助于识别瓶颈。

例如 - 如果您看到通常只有一条消息处于未确认状态，这可能意味着消费者无法处理该消息并将其发送回 rabbit。另一方面，新消息总是从生产者到达，所以 ready 消息的数量会增长得非常快 它还可以指出您只使用一个消费者一次只能处理一条消息。所以你可以在这里考虑并行化，通过在不同的线程中运行许多消费者甚至集群你的应用程序（在兔子消费者可以驻留在不同的机器上）

当然，正如我之前所说，如果您有更具体的问题，希望这总体上有所帮助 - 请提供有关测试期间究竟发生了什么的更多信息