使用 AMQP、Rabbit 和 Ruby 刷新队列

Question

我正在用 Ruby 开发一个系统，该系统将利用 RabbitMQ 在执行某些工作时将消息发送到队列。我正在使用：

• Ruby 1.9.1 Stable
• RabbitMQ 1.7.2
• AMQP gem v0.6.7 (http://github.com /tmm1/amqp）

我在这个 gem 上看到的大多数示例都在 EM.add_periodic_timer 块中进行发布调用。这不适用于我怀疑的绝大多数用例，当然也不适用于我的用例。我需要在完成一些工作时发布一条消息，因此将发布语句放在 add_periodic_timer 块中是不够的。

因此，我试图弄清楚如何将一些消息发布到队列，然后“刷新”它，以便我发布的任何消息都会传递给我的订阅者。

为了让您了解我的意思，请考虑以下发布者代码：

#!/usr/bin/ruby

require 'rubygems'
require 'mq'

MESSAGES = ["hello","goodbye","test"]

AMQP.start do

  queue = MQ.queue('testq')
  messages_published = 0

  while (messages_published < 50)

    if (rand() < 0.4)
      message = MESSAGES[rand(MESSAGES.size)]
      puts "#{Time.now.to_s}: Publishing: #{message}"
      queue.publish(message)
      messages_published += 1
    end

    sleep(0.1)

  end

  AMQP.stop do
    EM.stop
  end

end

因此，此代码只是循环，在循环的每次迭代中以 40% 的概率发布消息，然后休眠 0.1 秒。它会执行此操作，直到发布 50 条消息，然后停止 AMQP。当然，这只是一个概念证明。

现在，我的订阅者代码：

#!/usr/bin/ruby

require 'rubygems'
require 'mq'

AMQP.start do

  queue = MQ.queue('testq') 
  queue.subscribe do |header, msg|
    puts "#{Time.now.to_s}: Received #{msg}"
  end

end

因此，我们只需订阅队列，对于收到的每条消息，我们都会将其打印出来。

很好，只不过订阅者仅在发布者调用 AMQP.stop 时收到全部 50 条消息。

这是我的出版商的输出。为了简洁起见，它在中间被截断：

$ ruby publisher.rb 
2010-04-14 21:45:42 -0400: Publishing: test
2010-04-14 21:45:42 -0400: Publishing: hello
2010-04-14 21:45:42 -0400: Publishing: test
2010-04-14 21:45:43 -0400: Publishing: test
2010-04-14 21:45:44 -0400: Publishing: test
2010-04-14 21:45:44 -0400: Publishing: goodbye
2010-04-14 21:45:45 -0400: Publishing: goodbye
2010-04-14 21:45:45 -0400: Publishing: test
2010-04-14 21:45:45 -0400: Publishing: test
.
.
.
2010-04-14 21:45:55 -0400: Publishing: test
2010-04-14 21:45:55 -0400: Publishing: test
2010-04-14 21:45:55 -0400: Publishing: test
2010-04-14 21:45:55 -0400: Publishing: goodbye

接下来是我的订阅者的输出：

$ ruby consumer.rb
2010-04-14 21:45:56 -0400: Received test
2010-04-14 21:45:56 -0400: Received hello
2010-04-14 21:45:56 -0400: Received test
2010-04-14 21:45:56 -0400: Received test
2010-04-14 21:45:56 -0400: Received test
2010-04-14 21:45:56 -0400: Received goodbye
2010-04-14 21:45:56 -0400: Received goodbye
2010-04-14 21:45:56 -0400: Received test
2010-04-14 21:45:56 -0400: Received test
.
.
.
2010-04-14 21:45:56 -0400: Received test
2010-04-14 21:45:56 -0400: Received test
2010-04-14 21:45:56 -0400: Received test
2010-04-14 21:45:56 -0400: Received goodbye

如果您注意到输出中的时间戳，订阅者只会在发布者停止 AMQP 并退出后收到所有消息。

那么，作为 AMQP 新手，我怎样才能让我的消息立即传递呢？我尝试将 AMQP.start 和 AMQP.stop 放入发布者的 while 循环体中，但随后仅传递第一条消息 - 但奇怪的是，如果我打开日志记录，服务器不会报告任何错误消息，并且消息确实被发送到队列，但订阅者永远不会收到。

如有建议，我们将不胜感激。感谢您的阅读。

Answer 1

如果有人想要了解此问题的信息，请参阅 http://groups.google .com/group/ruby-amqp/browse_thread/thread/311965bcf9697ece

我通过在发布者代码中添加额外的线程解决了该问题：

!/usr/bin/ruby

require 'rubygems'
require 'mq'

MESSAGES = ["hello","goodbye","test"]

AMQP.start do

  Thread.new do
      queue = MQ.queue('testq')
      messages_published = 0

      while (messages_published < 50)

        if (rand() < 0.4)
          message = MESSAGES[rand(MESSAGES.size)]
          puts "#{Time.now.to_s}: Publishing: #{message}"
          queue.publish(message)
          messages_published += 1
        end

        sleep(0.1)

      end

      AMQP.stop do
        EM.stop
      end

  end

end