首页 文章

使用RabbitMQ可以放慢同步发布/消耗速度

提问于
浏览 次
5

我正在评估RabbitMQ，虽然（AMQP本身，以及RabbitMQ）的总体印象是积极的，但我对结果并不是很满意 .

我正在尝试同时发布和使用消息，并且已经实现了非常差的消息速率 . 我有一个持久的直接交换，它绑定到一个持久的队列，我向该交换发布持久性消息 . 消息正文的平均大小约为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) {}

并且从绑定队列中消费消息的方式如下：

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博客条目的一句话，声称队列在空闲时很快就可以，但是当队列中有几千条持久性消息时，消耗速度降低到爬行速度仍然是相当不可接受的 .

较高的QOS预取值可能有所帮助，但恕我直言不是这样的解决方案 .

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

rabbitmq amqp

2 回答

  • 5

    您是否尝试过autoAck选项？这应该可以改善你的表现 . 它比逐个获取消息并快速获取消息要快得多 . 增加预取计数应该会使它更好 .

    此外，您发送和消费的邮件大小包括 Headers ？您是否在经纪人中遇到任何流量控制？

    另一个问题是，每次发送/获取消息时，您是否创建了连接和通道？如果是这样，那就错了 . 您应该创建一次连接，并使用每个线程的通道（可能以线程本地方式）来发送和接收消息 . 每个连接可以有多个通道 . 没有关于此的官方文档，但如果你阅读文章和论坛，这似乎是最好的表现练习 .

    最后一件事，您是否考虑使用 basicConsume 而不是 basicGet ？它也应该使它更快 .

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

    回复于
  • -1

    如果使用 sleep ，操作系统可以将您的流程安排到下一个时间段 . 这可能会导致性能显着下降 .

    回复于

相关问题