EventMachine EM::Iterator 被rabbitmq RPC 阻塞

Question

我正在尝试设置 RabbitMQ rpc。我想要一个队列来侦听，当它收到一条消息时，我希望它回复通过带有多条消息的reply_to标头指定的匿名队列。

我有以下 thor 任务创建一个队列，然后使用 EM:Iterator 将许多消息发送回使用replyt_to路由键指定的队列：

desc "start_consumer", "start the test consumer"
def start_consumer
  conf = {
    :host => "localhost", 
    :user => "guest", 
    :password => "guest", 
    :vhost => "/", 
    :logging => true, 
    :port => 5672
  }

  # n = 1

  AMQP.start(conf) do |connection|

    channel = AMQP::Channel.new(connection)

    requests_queue = channel.queue("one")
    requests_queue.purge

    Signal.trap("INT") do
      connection.close do
        EM.stop{exit}
      end
    end

    channel.prefetch(1)

    requests_queue.subscribe(:ack => true) do |header, body|
      url_search = MultiJson.decode(body)

      EM::Iterator.new(0..5).each do |n, iter|
        lead = get_lead(n, (n == 5))

        puts "about to publish #{n} message is_last = #{lead.is_last} at #{Time.now}"

        AMQP::Exchange.default.publish(
                                        MultiJson.encode(lead), 
                                        :immediate => true,
                                        :routing_key => header.reply_to,
                                        :correlation_id => header.correlation_id
                                      )  

        iter.next
      end
    end

    puts " [x] Awaiting RPC requests"
  end          
end

下面的代码将一条消息发送到上面指定的队列，并创建一个队列，该队列将用于监听 EM::Iterator 代码发送的消息。该队列的名称是第一个队列的reply_to标头的路由键。

def publish(urlSearch, routing_key)
  EM.run do
    corr_id = rand(10_000_000).to_s

    requests ||= Hash.new

    connection = AMQP.connect(:host => "localhost")

    callback_queue = AMQP::Channel.new(connection).queue("", :exclusive => false)

    callback_queue.subscribe do |header, body|  
      lead = safe_json_decode(body)

      puts "company = #{lead["company"]} is_last = #{lead["is_last"]} received at #{Time.now}"              

      if lead["is_last"]
        puts "in exit"
        connection.close do
          EM.stop{exit}
        end
      end
    end

    callback_queue.append_callback(:declare) do
      AMQP::Exchange.default.publish(MultiJson.encode(urlSearch), :routing_key => routing_key, :reply_to => callback_queue.name, :correlation_id => corr_id)
    end

    puts "initial message sent"
  end
end

上面的代码按照我想要的方式工作，但有一个令人讨厌的异常。有些东西阻止了 EM::Iterator 代码的异步执行。只有在 EM::Iterator 代码完成后才会发送消息。我希望消息异步发送，并在每次迭代后由匿名队列处理。目前，只有在 EM::Iterator 代码完成最后一次迭代之后，才会发送所有消息。

任何人都可以看到我做错了什么或建议不同的方法吗？我尝试了 EM::defer 并有相同的行为。

Answer 1

启动一个新线程是我问题的答案：

Thread.new do
  5.times do
    lead = get_lead(n, (n == 5))

    puts "message #{n} is_last = #{lead.is_last} at #{Time.now}";

    AMQP::Exchange.default.publish(
                                    MultiJson.encode(lead), 
                                    :routing_key => header.reply_to,
                                    :correlation_id => header.correlation_id
                                  )

    n += 1
    sleep(2)
  end
end

创建一个新线程会阻止 EventMachine 反应器被阻塞，并且消息会异步发送。