0

我正在处理来自我的 rabbitMQ 的记录。为了避免来自rabbitMQ的消息丢失，我正在使用rabbitMQ确认策略。出于测试目的，我在我向 rabbitMQ 回馈确认的代码之前提出了异常。根据我编写的代码 rabbitMQ 不应该从队列中删除消息，直到我的代码中的行，即“c.ack（delivery_info.delivery_tag）”被执行。下面是我的代码

begin
MessageService.with_channel do |c|  
  queue = c.queue(QUEUE_NAME, :durable => true)
  exchange = c.topic(TOPIC_NAME, :durable => true)
  queue.bind(exchange, :routing_key => "#{ROUTING_KEY_PREFIX}.#")
  available_messages = queue.message_count
  processed_messages = 0
  fact_object = {}
  while queue.message_count > 0 && processed_messages <= message_limit
    queue.pop(:manual_ack => true) do |delivery_info, properties, payload|
      rad_exam_hash_object = JSON.parse(payload)
      entity_manager,criteria = self.query_revenue_cost_calculation(rad_exam_hash_object)
      fact_object = self.revenue_cost_calculation_result(entity_manager,criteria)
      self.insert_data_into_table(fact_object)
      raise "exception"
      c.ack(delivery_info.delivery_tag)
    end
    processed_messages = processed_messages + 1
    available_messages = processed_messages - 1
  end
  [processed_messages, available_messages,fact_object]
end
rescue
  puts "got exception while processing messages"
end

请帮助我理解为什么即使“c.ack(delivery_info.delivery_tag)”行没有执行，rabbitMQ 也会从队列中删除消息。

4

1 回答 1

0

免责声明：我不知道红宝石。消息从队列中删除：

  • 如果队列被删除
  • 如果他们被承认
  • 如果消息 TTL 过期。

因此，如果不是这样，也许它的pop工作方式。这听起来像是在调用 AMQPsget并且（基于教程）subscribe似乎在调用 AMQPs consume。

于 2016-10-26T11:04:24.810 回答