【新手求学】Nodejs+rabbitmq 使用amqp（或者amqplib问题一样） 出现瓶颈求解

【新手求学】Nodejs + RabbitMQ 使用 amqp（或者amqplib问题一样） 出现瓶颈求解

【内容描述】

使用 Node.js 作为生产者将消息发布到位于另一台计算机上的 RabbitMQ 服务器。

【目标】

发送一个 4KB 的文本消息，最短时间发送 10 万个消息。

【现象】

1. 当消息大小为 4KB，并且一次性通过 for 循环发布 4 万个消息时，速度很快，CPU 使用率较低（仅在 for 循环期间较高，但时间很短）。然而，超过 5 万个消息后，速度显著下降，CPU 使用率显著升高，且需要很长时间才能完成。只有当剩余的消息数量降至 4 万个以内时，速度才会提升，CPU 使用率下降。
2. 当消息较小（只有几个字节）时，即使循环发布 10 万个或 20 万个消息，速度也很快。
3. 如果 RabbitMQ 服务器与 Node.js 生产者在同一台机器上，则无论消息大小如何，都不会出现上述问题，速度也会快得多。

【我的问题】

为什么当消息大小为 4KB 并且 RabbitMQ 服务器位于远程机器上时，会出现一个大约 5 万个消息的界限？如何解决这个问题？

【解决方案】

该问题主要由于网络延迟和 RabbitMQ 服务器处理大量消息的能力限制所致。为了解决此问题，可以采取以下措施：

1. 异步发布消息：确保消息发布操作是异步的，以避免阻塞事件循环。
2. 批量发布消息：而不是一次性发布所有消息，可以分批次发布消息，减少每次请求的数据量。

示例代码

const amqp = require('amqp');
const fs = require('fs');

// 配置 RabbitMQ 连接
const connOptions = {
    host: '192.168.65.170',
    port: 5672,
    login: 'guest',
    password: 'guest',
    authMechanism: 'AMQPLAIN',
    vhost: '/',
    ssl: {
        enabled: false
    }
};

// 定义交换机选项
const exchOption = {
    type: 'topic',
    durable: true,
    autoDelete: false,
    confirm: false
};

// 定义消息选项
const messOption = {
    contentEncoding: "utf-8",
    deliveryMode: 1
};

// 读取消息内容
const message = fs.readFileSync('2.txt').toString('utf-8');
const n = 100000; // 循环次数

// 异步发布消息
function publishMessage(exchange) {
    const batchSize = 1000;
    let count = 0;

    function publishBatch() {
        for (let i = 0; i < batchSize && count < n; i++) {
            exchange.publish('topic', new Buffer(message), '', () => {
                count++;
                if (count % batchSize === 0 || count === n) {
                    console.log(`Published ${count} messages`);
                }
            });
        }

        if (count < n) {
            setTimeout(publishBatch, 0); // 延迟一段时间继续发布下一批
        }
    }

    publishBatch();
}

// 连接 RabbitMQ 并创建交换机
const conn = amqp.createConnection(connOptions);

conn.on('ready', () => {
    console.log("RabbitMQ connection ready");
    const exch = conn.exchange('topic', exchOption, publishMessage);
});

解释

1. 异步发布消息：通过 setTimeout 实现异步发布，避免阻塞事件循环。
2. 批量发布消息：每次发布 1000 条消息，然后等待下一批次发布。这有助于减轻网络负载并提高整体性能。