将大量消息发布到AMQP队列

Question

使用 Ruby amqp 库的 v0.7.1 和 Ruby 1.8.7，我尝试将大量（数百万）短（~40 字节）消息发布到 RabbitMQ 服务器。我的程序的主循环（好吧，不是真正的循环，但仍然）看起来像这样：

AMQP.start(:host => '1.2.3.4', 
       :username => 'foo',
       :password => 'bar') do |connection|

  channel  = AMQP::Channel.new(connection)
  exchange = channel.topic("foobar", {:durable => true})
  i = 0

  EM.add_periodic_timer(1) do
    print "\rPublished #{i} commits"
  end

  results = get_results # <- Returns an array 

  processor = proc do
    if x = results.shift then
        exchange.publish(x, :persistent => true, 
                         :routing_key => "test.#{i}")
        i += 1
        EM.next_tick processor
      end
  end
  EM.next_tick(processor)
  AMQP.stop {EM.stop} end

代码开始处理结果数组很好，但过了一会儿（通常是在 12k 消息之后左右）它就会终止，并出现以下错误

/Library/Ruby/Gems/1.8/gems/amqp-0.7.1/lib/amqp/channel.rb:807:in `send': 
The channel 1 was closed, you can't use it anymore! (AMQP::ChannelClosedError)

No messages都存储在队列中。该错误似乎是在从程序到队列服务器的网络活动开始时发生的。

我做错了什么？

Answer 1

第一个错误是您没有发布您正在使用的 RabbitMQ 版本。许多人正在运行旧的过时版本 1.7.2，因为这是他们的操作系统包存储库中的版本。对于像您这样发送大量消息的人来说，这是一个糟糕的举动。从 RabbitMQ 站点本身获取 RabbitMQ 2.5.1 并删除默认的系统包。

第二个错误是您没有告诉我们 RabbitMQ 日志中的内容。

第三个错误是您没有提及消耗消息的内容。是否有另一个进程在某处运行，该进程已声明队列并将其绑定到交换器。除非有人将消息队列声明给 RabbitMQ 并将其绑定到交换器，否则没有消息队列。即使这样，只有当队列的绑定键与您发布的路由键匹配时，消息才会流动。

第四个错误。您混淆了路由密钥和绑定密钥。路由键是一个字符串，例如 topic.test.json.echos，绑定键（用于将队列绑定到交换器）是一个模式，例如 topic.# 或 topic..json。

< em>在您澄清后更新
关于版本，我不确定它何时被修复，但 1.7.2 中存在一个问题，大量持久消息导致 RabbitMQ 在滚动其持久日志时崩溃，并且崩溃后无法重新启动，直到有人手动取消翻转。

当您说正在打开和关闭连接时，我希望它不是每条消息。这将是一种奇怪的使用 AMQP 的方式。

让我重复一遍。生产者不将消息写入队列。它们将消息写入交换器，然后交换器根据路由键（字符串）和队列的绑定键（模式）将消息路由到队列。在您的示例中，我误读了 # 符号的使用，但我没有看到任何声明队列并将其绑定到交换器的内容。