Open pandatools opened 1 month ago
上面是测试代码,复现方式,断掉电脑wifi,过一到两分钟,等他出现RabbitmqStore consume connection error<Connection dead, no heartbeat or data received in >= 60s> reconnecting...,再重新连接wifi,即可复现出问题
问题现象:当重连后,不会消费新数据
def start_consuming(
self, queue_name: str, callback: Callable, prefetch=1, **kwargs
):
"""替代原版start_consuming 该现象可以解决"""
self.__shutdown = False
no_ack = kwargs.pop("no_ack", False)
reconnection_delay = self.RECONNECTION_DELAY
while not self.__shutdown:
try:
if self.isok: #init 函数开始默认为true
channel = self.channel
logger.info('use old')
else:
parameters = {
"confirm_delivery": True,
"hostname": "xxxx",
"port": 5672,
"username": "xxx",
"password": "xxxx",
"ssl": False
}
logger.info('use new')
connection = amqpstorm.Connection(**parameters)
channel = connection.channel()
channel.confirm_deliveries()
channel.basic.qos(prefetch_count=prefetch)
channel.basic.consume(
queue=queue_name, callback=callback, no_ack=no_ack, **kwargs
)
channel.start_consuming(to_tuple=False)
self.isok = True
except AMQPConnectionError as exc:
logger.warning(
f"RabbitmqStore consume connection error<{exc}> reconnecting..."
)
del self.connection
self.isok = False
reconnection_delay = min(
reconnection_delay * 2, self.MAX_CONNECTION_DELAY
)
except Exception as e:
if self.__shutdown:
break
logger.exception(f"RabbitmqStore consume error<{e}>, reconnecting...")
del self.connection
time.sleep(reconnection_delay)
reconnection_delay = min(
reconnection_delay * 2, self.MAX_CONNECTION_DELAY
)
finally:
if self.__shutdown:
break``