刘立权
RabbitMQ死循环-延长ACK时间
来源:刘勇     发布时间: 2019-08-21      浏览次数:461

字号:

一、应用背景

  今天做一个需求,要将RabbitMQ中的任务取出并执行,为防止任务执行期间出错,设置NO_ACK=FALSE标志,这样、一旦任务没有应答的话,相应的任务就会被RabbitMQ自动Re-Queue,避免丢失任务。然而、由于任务执行时间较长,通常需要五、六分钟,甚至更长;我们都知道一旦一个任务被取出执行,该任务就从Ready状态更改成Unacked状态。如图所示:

  当这个任务执行完之后,程序将向RabbitMQ发送ACK消息确认,RabbitMQ在收到ACK消息后,会将该任务移出队列;然而、问题出在任务尚未执行完毕【执行时间太久】,RabbitMQ再等了一段时间【大约两三分钟】后,一直没有收到ACK确认消息,就将该任务自动Re-Queue了【我是一个生产者,一个消费者模式】,也就是说、我们这里发生了死循环【任务永远也执行不完,因为会一直Re-Queue】。

二、延长RabbitMQ ACK应答时间

  到这里,我们急需解决的问题就是,怎么能设置RabbitMQ延长等待ACK的时间,百度一下、两下,各种读网络文档,研究操作RabbitMQ工作的文档,查了一圈资料也没查出怎么延长RabbitMQ ACK时间【废柴啊】。至此、一直查不出来,就想问一下网友的你,你知道怎么延长RabbitMQ接受ACK应答时间么?

三、改变解决问题方式

  在查不出如何延长ACK应答时间后,我将注意力转向如何检测当前任务操作超时的,后来在官网看到这么一段话:

  链接官网位置:http://www.rabbitmq.com/heartbeats.html#heartbeats-timeout

  

   后面、就简单测试下将heartbeat参数设置为0,以禁用心跳检测,这样基本解决了我的问题;虽然官方不建议这么做,但也是一种解决思路,如果大家有什么更好的解决办法,烦请在下面留言【先谢谢啦】。

  至此、这个问题基本阐述清楚了,如果有遇到的小伙伴,也请参考下上面的操作。

  测试代码:

 

# import json# from concurrent.futures import ThreadPoolExecutorfrom queue import Queue# from threading import Threadfrom pika import BasicProperties, BlockingConnection, URLParametersfrom pika.exceptions import ConnectionClosed# from automation.aiclient.aiclient import AsyncAIRequestManagerclass RabbitMQManager: def __init__(self, host = "localhost", qname = "queue"): self.params = URLParameters(host) self.qname = qname self.prod_conn = None self.prod_chan = None self.cons_conn = None self.cons_chan = None self.ai_signton = None def init_prod_conn(self): # create send connection self.prod_conn = BlockingConnection(self.params) self.prod_chan = self.prod_conn.channel() self.prod_chan.queue_declare(queue = self.qname, durable = True) def init_cons_conn(self): # create receive connection self.cons_conn = BlockingConnection(self.params) self.cons_chan = self.cons_conn.channel() self.cons_chan.basic_qos(prefetch_count = 1) self.cons_chan.basic_consume(self.callback, queue = self.qname) def produceMessages(self, msg = None): try: if isinstance(msg, str): self.prod_chan.basic_publish(exchange = "", routing_key = self.qname, body = msg, properties = BasicProperties( delivery_mode = 2, # make message persistent )) elif isinstance(msg, Queue): while 0 != msg.qsize(): item = msg.get() self.prod_chan.basic_publish(exchange = "", routing_key = self.qname, body = item, properties = BasicProperties( delivery_mode = 2, # make message persistent )) else: pass except Exception as e: if isinstance(e, ConnectionClosed): print("Reconnection established!") self.init_prod_conn() # last connection close, re-produce msg self.produceMessages(msg) else: print("Produce msg exception Occur, please check following error message:") print(e) def consumeMessages(self): try: self.cons_chan.start_consuming() except Exception as e: print("Consume msg exception Occur, please check following error message:") print(e) if isinstance(e, ConnectionClosed): print("Reconnection established!") self.init_cons_conn() self.consumeMessages() def callback(self, ch, method, properties, body): # handle message body print("callback....") print(body) try: print("Consuming....") self.cons_conn.process_data_events() # 模拟处理任务时间 import time time.sleep(300) # if None == self.ai_signton: # self.ai_signton = AsyncAIRequestManager() # self.ai_signton.run(eval(json.loads(json.dumps(body.decode("utf-8")), encoding = "utf-8"))) ch.basic_ack(delivery_tag = method.delivery_tag) # t = Thread(target = self.ai_signton.syncToDatabase()) # t.start() except Exception as e: if isinstance(e, ConnectionClosed): raise ConnectionClosed("Connection has been closed, send to reconnection.") else: print("Current error msg:") print(e) def close_prod_conn(self): if None != self.prod_conn: self.prod_conn.close() def close_cons_conn(self): if None != self.cons_conn: self.cons_conn.close() def close(self): self.close_prod_conn() self.close_cons_conn()

  

  • 相关内容: