ruby 如何将消息从unacked推送到ready

vqlkdk9b  于 12个月前  发布在  Ruby
关注(0)|答案(2)|浏览(82)

我的问题与之前提出的问题类似,但是它没有找到答案,我有一个消费者,我想处理一个名为Web服务的操作,但是,如果这个Web服务由于某种原因没有响应,我希望消费者不要处理RabbitMQ的消息,但我会让它稍后处理它,我的消费者是下面的一个:

require File.expand_path('../config/environment.rb', __FILE__)
conn=Rabbit.connect
conn.start
ch = conn.create_channel
x = ch.exchange("d_notification_ex", :type=> "x-delayed-message", :arguments=> { "x-delayed-type" => "direct"})
q  = ch.queue("d_notification_q", :durable =>true)
q.bind(x)
p 'Wait ....'
q.subscribe(:manual_ack => true, :block => true) do |delivery_info, properties, body|
 
  datos=JSON.parse(body)
  if datos['status']=='request'
    #I call a web service and process the json
    result=Notification.send_payment_notification(datos.to_json)
  else
    #I call a web service and process the body
    result=Notification.send_payment_notification(body)
  end
   #if the call to the web service, the web server is off the result will be equal to nil
   #therefore, he did not notify RabbitMQ, but he puts the message in UNACKED status
   # and does not process it later, when I want him to keep it in the queue and evaluate it afterwards.
  unless result.nil?
  ch.ack(delivery_info.delivery_tag)
  end

end

RabbitMQ的图片,

在声明中有某种方式:c hack(delivery_info.delivery_tag),这不是删除队列中的元素,可以稍后再处理,有什么想法吗?谢谢

vlurs2pr

vlurs2pr1#

我决定用“消费者中的生产者”的样式将数据发送回队列,我的代码现在看起来像这样:

if result.eql? 'ok'
  ch.ack(delivery_info.delivery_tag)
  else
    if(datos['count'] < 5)
      datos['count'] += 1
      d_time=1000
      x.publish(datos.to_json,  :persistent => true, :headers=>{"x-delay" => d_time})
    end
  end

然而,我不得不在JSON属性中添加一个属性:计数!这样它就不会停留在无限循环中。

kse8i1jr

kse8i1jr2#

试试这个:

if result.nil?
  ch.nack(delivery_info.delivery_tag)
else
  ch.ack(delivery_info.delivery_tag)
end

相关问题