rabbitmq - 使用 RabbitMQ 的同时发布/消费速度慢得离谱

Question

我正在评估 RabbitMQ，虽然(对 AMQP 本身以及 RabbitMQ)的总体印象是积极的，但我对结果印象不深。

我正在尝试同时发布和使用消息，但消息率非常低。我有一个持久的直接交换，它绑定(bind)到一个持久队列，我将持久消息发布到该交换。消息正文的平均大小约为 1000 字节。

我的发布大致如下:

AMQP.BasicProperties.Builder bldr = new AMQP.BasicProperties.Builder();
ConnectionFactory factory = new ConnectionFactory();
factory.setUsername("guest");
factory.setPassword("guest");
factory.setVirtualHost("/");
factory.setHost("my-host");
factory.setPort(5672);
Connection conn = null;
Channel channel = null;
ObjectMapper mapper = new ObjectMapper(); //com.fasterxml.jackson.databind.ObjectMapper
try {
    conn = factory.newConnection();
channel = conn.createChannel();
    channel.confirmSelect();
} catch (IOException e) {}

for(Message m : messageList) { //the size of messageList happens to be 9945
    try {
        channel.basicPublish("exchange", "", bldr.deliveryMode(2).contentType("application/json").build(), mapper.writeValueAsBytes(cm));
    } catch (Exception e) {}
}
try {
    channel.waitForConfirms();
    channel.close();
conn.close();
} catch (Exception e1) {}               

并且从绑定(bind)队列中消费消息是这样发生的:

AMQP.BasicProperties.Builder bldr = new AMQP.BasicProperties.Builder();
ConnectionFactory factory = new ConnectionFactory();
factory.setUsername("guest");
factory.setPassword("guest");
factory.setVirtualHost("/");
factory.setHost("my-host");
factory.setPort(5672);
Connection conn = null;
Channel channel = null;
try {
    conn = factory.newConnection();
    channel = conn.createChannel();
    channel.basicQos(100);
    while (true) {
        GetResponse r = channel.basicGet("rawDataQueue", false);
        if(r!=null)
            channel.basicAck(r.getEnvelope().getDeliveryTag(), false);
    }
} catch (IOException e) {}

问题是，当消息发布者(或其中几个)和消费者(或其中几个)同时运行时，发布者似乎全速运行，并且 RabbitMQ 管理 Web 界面显示的发布速率为每秒约 2...3K 条消息，但每个消费者的消费率为 0.5...3。当发布者完成时，我得到的消费率是每个消费者 300...600 条消息。不为Java客户端设置QOS预取值时，则少一点，设置为100或250时，则多一点。

在尝试对消费者进行一定程度的限制时，我设法同时实现了每秒大约 400 条已发布消息和大约 50 条消费消息，这稍微好一些，但也只是勉强。

Here's ，来自 RabbitMQ 博客条目的引用声称队列在空时最快，这很可能是，但是当队列中有几千条持久消息时，将消耗率减慢到爬行仍然是相当 Not Acceptable 。

更高的 QOS 预取值可能会有所帮助，但恕我直言，这不是一个解决方案。

如果有的话，可以做些什么来实现合理的吞吐率(在任何情况下，每个消费者每秒消耗 2 条消息都是不合理的)？这只是一个简单的一个直接交换 - 一个绑定(bind) - 一个队列的情况，我是否应该期望更复杂的配置会导致更多的性能下降？在互联网上搜索时，也有人建议降低耐用性，但恐怕在我的情况下这不是一个选择。如果有人指出我很愚蠢并且有某种明显而直接的解决方案，我会很高兴:)

Answer 1

您是否尝试过使用 autoAck 选项？这应该会提高你的表现。它比一个一个地获取消息并确认它们要快得多。增加预取计数也应该使它变得更好。

此外，您发送和使用的消息(包括 header )的大小是多少？您是否在代理中遇到任何流量控制？

另一个问题，您是否在每次发送/接收消息时都创建连接和 channel ？如果是这样，那就错了。您应该创建一次连接，并使用每个线程的 channel (可能以线程本地方式)来发送和接收消息。每个连接可以有多个 channel 。没有关于此的官方文档，但如果您阅读文章和论坛，这似乎是最佳性能实践。

最后一件事，您是否考虑过使用 basicConsume而不是 basicGet ?它还应该使它更快。

根据我的经验，我已经能够运行集群以每秒大约 20000 条消息的速率发送和使用非持久消息。我想如果你使用持久和持久的消息，性能会下降一点，但不会下降 10 倍。