2

我有一个使用 RabbitMQ 和 Sneakers gem 的用例，我有多个 Worker 运行以响应我项目中的几十个队列。因此，工作人员很可能同时处理来自同一个队列的消息。

但是，特别是对于一个队列 - 让我们称之为:one_at_a_time- 我只希望一个工作人员能够在任何给定时间处理来自队列的消息。

我想这样做的原因是因为工作人员旨在执行以下操作：

  1. 通过传入的 ID 查找 AR 对象
  2. 检查是否设置了一个属性——比如 :worked。
    1. 如果true，则ack!消息。
    2. 如果false，向用户发送电子邮件，则设置:worked为 true。

这样做是为了防止我不小心向用户发送两次电子邮件，以防两个消息以相同的对象 ID 快速连续创建。如果在任何给定时间只有一个工作人员听过这个队列，这个设计就可以正常工作，因为第一次运行将经过步骤 1 -> 2 -> 2，然后下一次运行将经过步骤 1 -> 2 -> 1 并且不会向用户发送电子邮件。但在测试中，我发现有可能出现竞争条件，即两个工作人员会:one_at_a_time同时从队列中拉出一条消息，通过:worked设置的检查，然后都发送一封电子邮件。

所以考虑到这一切，有没有办法可以限制收听队列的工人数量？谢谢你。

4

1 回答 1

1

如需进一步参考，可以按照以下步骤归档 Argus9 的请求：

1) 你可以控制你的工人的选择是：

class YourWorker
include Sneakers::Worker
from_queue "your_queue",
           :env => nil,
           :ack => true,
           :workers => 1, #Number of per-cpu processes to run
           :prefetch => 1, #This param will define that single message will be fetched per time
           :threads => 1, #This will define that you have single thread running
           :heartbeat => 2,
           :share_threads => true,
           :timeout_job_after => 3600,
           :exchange => 'your_exchange'

def work(args={})
 #... your steps here
end 
end

2）你需要注意你在你的sneakers.rb中指定的初始参数（因为它被Sneakers::Runner在worker初始化时使用），所以一定要在那里有正确的参数，比如：

Sneakers.configure  :amqp => url,
                :daemonize => true,
                :ack => true,
                :prefetch => 1,
                :threads => 1,
                :start_worker_delay => 0.1,  
                :workers => 1,               
                :exchange => "your_exchange",
                :exchange_type => :direct,
                :log => "log/sneakers.log"
Sneakers.logger.level = Logger::DEBUG

您还可以使用 RabbitMQ API 构建一些额外的控件，这将使您能够检查诸如是否已经有一些消息正在处理？...等等使用 bunny 归档并不容易等等。使用非常简单的代码，例如：

    def queue_info
    queues_infos = {}    
    rabbitmqctl_url = "http://127.0.0.1:15672"
    rabbitmqctl_user = "your_user"
    rabbitmqctl_password = "your_password"
    uri = URI.parse("#{rabbitmqctl_url}/api/queues")
    request = Net::HTTP::Get.new(uri)
    request.basic_auth(rabbitmqctl_user, rabbitmqctl_password)
    req_options = { use_ssl: uri.scheme == 'https' }
    response = Net::HTTP.start(uri.hostname, uri.port, req_options)  do |http|
      http.request(request)
    end
    queue_details = JSON.parse(response.body)
    queue_details.each do |queue|
      queues_infos[queue['name'].to_s] = {  name: queue['name'],
                                            msg_total: queue['messages'],
                                            msg_ready: queue['messages_ready'],
                                            msg_unacknowlged: queue['messages_unacknowledged'],
                                            state: queue['state'],
                                            consumers: queue['consumers'] }
    end
    return queues_infos
end
于 2018-10-30T16:48:14.967 回答