0

使用 Ruby 的“bunny”RabbitMQ 客户端，我希望我的生产者（一些 Ruby 代码）向消费者（使用“sneakers”gem 的工作人员）发送一条消息，并且我希望我的生产者不执行其 Ruby 代码的另一行，直到生产者收到消费者收到我的消息并对其进行处理的确认。

在我的消费者中，我正在做一些工作，然后调用sneakers 的ack!方法来确认消息已收到并且工作已完成。

在我的生产者中，我正在调用confirm_select我的Bunny::Channel实例将其放入确认代码中，并且在publish我的消息之后，我调用wait_for_confirms通道以等待我的所有消息都被ack!消费者 -ed。（我已经尝试实现我在 bunny docs here中找到的内容。）

但是，似乎我的生产者并没有等待消费者调用ack!。我正在登录我的生产者和消费者，并发现我的生产者似乎认为消息在消费者实际确认之前已被确认。

如何让 RabbitMQ 生产者等到消费者在 Ruby 中完成工作？

Ruby 2.3.3、RabbitMQ 3.6.12、Erlang 17.3。

这是我的锁文件：

GEM
  specs:
    amq-protocol (2.2.0)
    bunny (2.7.0)
      amq-protocol (>= 2.2.0)
    concurrent-ruby (1.0.5)
    serverengine (1.5.11)
      sigdump (~> 0.2.2)
    sigdump (0.2.4)
    sneakers (2.6.0)
      bunny (~> 2.7.0)
      concurrent-ruby (~> 1.0)
      serverengine (~> 1.5.11)
      thor
    thor (0.20.0)

PLATFORMS
  ruby

DEPENDENCIES
  bunny
  sneakers

BUNDLED WITH
   1.14.6

这是我的消费者/工人（consumer_worker.rb）：

class ConsumerWorker
  include Sneakers::Worker

  from_queue 'do-work-here',
             exchange: 'do-work-here',
             exchange_type: :direct,
             durable: true,
             prefetch: 1,
             arguments: {
               :'x-dead-letter-exchange' => 'do-work-here-retry'
             },
             timeout_job_after: 5,
             retry_timeout: 60000,
             ack: true

  def work(msg)
    open('ruby-debug.log', 'a') do |f|
      f.puts "message received: #{msg}"
    end
    sleep 1
    open('ruby-debug.log', 'a') do |f|
      f.puts "acknowledging message at: #{Time.now.to_i}"
    end
    ack!
    open('ruby-debug.log', 'a') do |f|
      f.puts "acknowledged message at: #{Time.now.to_i}"
    end
  end
end

在一个终端选项卡中，我正在运行这个工作人员：

bundle exec sneakers work ConsumerWorker --require consumer_worker.rb

这是我的发布者（publisher.rb）：

require 'bunny'
connection = Bunny.new('amqp://guest:guest@localhost:5672').tap(&:start)
channel = connection.create_channel
channel.confirm_select
queue = channel.queue('do-work-here',
                      {arguments: {:'x-dead-letter-exchange' => 'do-work-here-retry'},
                       durable: true})
queue.publish('hello world', persistent: true)
channel.wait_for_confirms
open('ruby-debug.log', 'a') do |f|
  f.puts "messages confirmed at: #{Time.now.to_i}"
end

当我在另一个选项卡中运行以下命令时：

ruby ./publisher.rb

然后我的日志文件 ( ./ruby-debug.log) 包含以下几行：

message received: hello world
messages confirmed at: 1505774819
acknowledging message at: 1505774820
acknowledged message at: 1505774820

我想要的是事件的顺序是这样的：

message received
acknowledging message
acknowledged message
messages confirmed

我该怎么做？

4

1 回答 1

0

Publisher 确认仅涵盖发布者到 RabbitMQ 的通信。出版商不了解消费者。

有关请求/响应模式的示例，请参见教程 6： https ://www.rabbitmq.com/getstarted.html 。

ConditionVariable 是一种常用的并发原语，用于延迟进一步的操作，直到事件发生，并带有可选的超时。

出版商确认和消费者确认记录在http://www.rabbitmq.com/confirms.htm。

于 2017-09-18T23:44:50.873 回答