我填满我的队列,检查它是否有正确的任务数要工作,然后将parallell中的工作人员设置为prefetch(1),以确保每个人一次只处理一项任务。
我希望每个工作人员完成各自的任务,发送手动确认,如果有更多工作,则继续从队列中获取工作。
如果没有更多的工作,即队列为空,我希望worker脚本完成并执行return(0)。
所以,这是我现在所拥有的:
require 'bunny'
connection = Bunny.new("amqp://my_conn")
connection.start
channel = connection.create_channel
queue = channel.queue('my_queue_name')
channel.prefetch(1)
puts ' [*] Waiting for messages.'
begin
payload = 'init'
until queue.message_count == 0
puts "worker working queue length is #{queue.message_count}"
_delivery_info, _properties, payload = queue.pop
unless payload.nil?
puts " [x] Received #{payload}"
raise "payload invalid" unless payload[/cucumber/]
begin
do_stuff(payload)
rescue => e
puts "Error running #{payload}: #{e.backtrace.join('\n')}"
#failing stuff
end
end
puts " [x] Done with #{payload}"
end
puts "done with queue"
connection.close
exit(0)
ensure
connection.close
end 当队列为空时,我仍然希望确保完成。这是来自RabbitMQ站点的示例...https://www.rabbitmq.com/tutorials/tutorial-two-ruby.html。它有许多我们想要用于工作队列的东西,最重要的是手动确认。但它不会停止运行,我需要在队列完成时以编程方式停止运行:
#!/usr/bin/env ruby
require 'bunny'
connection = Bunny.new(automatically_recover: false)
connection.start
channel = connection.create_channel
queue = channel.queue('task_queue', durable: true)
channel.prefetch(1)
puts ' [*] Waiting for messages. To exit press CTRL+C'
begin
queue.subscribe(manual_ack: true, block: true) do |delivery_info, _properties, body|
puts " [x] Received '#{body}'"
# imitate some work
sleep body.count('.').to_i
puts ' [x] Done'
channel.ack(delivery_info.delivery_tag)
end
rescue Interrupt => _
connection.close
end当队列已经完全工作(总共0个,未确认0个)时,如何调整此脚本以退出?
发布于 2018-10-16 03:05:18
据我所知,如果RabbitMQ队列中没有挂起的消息,您希望您的订阅者结束订阅。
对于第二个脚本,您可以避免传递block: true,当没有更多的数据需要处理时,这将不会返回任何内容。在这种情况下,您可以退出程序。
可以在文档中看到这一点:http://rubybunny.info/articles/queues.html#blocking_or_nonblocking_behavior
默认情况下,它是非阻塞的。
https://stackoverflow.com/questions/52668224
复制相似问题