我正在建立一个分析系统，有一百万用户同时在线。我使用RabbitMQ如消息代理来减少服务器的容量
这是我的图表
x1c 0d1x的数据
我的系统包括3个组件。

**发布服务器：（Producer）**本系统基于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"交换机接收消息
当我实现发送消息时。系统运行良好，但是我尝试了这个系统的基准测试性能（每秒发送约1000个用户请求），然后系统出现了一些问题。系统似乎是超载/缓冲区溢出（或一些东西不工作）。
我两天前才读到rabbitmq。我知道它的教程是基本的例子，所以我需要帮助，以建立在真实的世界中的系统比。任何解决方案和建议
希望我的问题有意义