我有一个异步脚本，该脚本每40秒通过aio_pika库连接到Rabbitmq，并检查是否有任何消息并将其打印出来，然后永久重复。但是，通常，在运行约两天后，我将开始收到无尽的连接异常错误，这些错误只能通过重新启动脚本来解决。也许我缺少的异步脚本逻辑中存在一些明显的错误？
#!/usr/bin/python3
import time
import async_timeout
import asyncio
import aio_pika
async def got_message(message: aio_pika.IncomingMessage):
with message.process():
print(message.body.decode())
async def main(loop):
try:
with async_timeout.timeout(10):
connection = await aio_pika.connect_robust(
host='#',
virtualhost='#',
login='#',
password='#',
port=5671,
loop=loop,
ssl=True
)
channel = await connection.channel()
await channel.set_qos(prefetch_count=100)
queue_name='mm_message'
queue = await channel.declare_queue(auto_delete=False, name=queue_name)
routing_key='mm_msg'
await queue.bind("amq.topic", routing_key)
que_len = queue.declaration_result.message_count
if(que_len > 0):
await queue.consume(got_message)
except:
print("connection problems..")
if __name__ == "__main__":
loop = asyncio.get_event_loop()
while(True):
time.sleep(40)
loop.run_until_complete(main(loop))
这是一段时间后我不断收到的错误：
Traceback (most recent call last):
File "/usr/lib/python3.5/asyncio/events.py", line 125, in _run
self._callback(*self._args)
File "/usr/local/lib/python3.5/dist-packages/aio_pika/pika/adapters/base_connection.py", line 364, in _handle_events
self._handle_read()
File "/usr/local/lib/python3.5/dist-packages/aio_pika/pika/adapters/base_connection.py", line 415, in _handle_read
self._on_data_available(data)
File "/usr/local/lib/python3.5/dist-packages/aio_pika/pika/connection.py", line 1347, in _on_data_available
self._process_frame(frame_value)
File "/usr/local/lib/python3.5/dist-packages/aio_pika/pika/connection.py", line 1414, in _process_frame
if self._process_callbacks(frame_value):
File "/usr/local/lib/python3.5/dist-packages/aio_pika/pika/connection.py", line 1384, in _process_callbacks
frame_value) # Args
File "/usr/local/lib/python3.5/dist-packages/aio_pika/pika/callback.py", line 60, in wrapper
return function(*tuple(args), **kwargs)
File "/usr/local/lib/python3.5/dist-packages/aio_pika/pika/callback.py", line 92, in wrapper
return function(*args, **kwargs)
File "/usr/local/lib/python3.5/dist-packages/aio_pika/pika/callback.py", line 236, in process
callback(*args, **keywords)
File "/usr/local/lib/python3.5/dist-packages/aio_pika/pika/connection.py", line 1332, in _on_connection_tune
self._send_connection_open()
File "/usr/local/lib/python3.5/dist-packages/aio_pika/pika/connection.py", line 1517, in _send_connection_open
self._on_connection_open, [spec.Connection.OpenOk])
File "/usr/local/lib/python3.5/dist-packages/aio_pika/pika/connection.py", line 1501, in _rpc
self._send_method(channel_number, method_frame)
File "/usr/local/lib/python3.5/dist-packages/aio_pika/pika/connection.py", line 1569, in _send_method
self._send_frame(frame.Method(channel_number, method_frame))
File "/usr/local/lib/python3.5/dist-packages/aio_pika/pika/connection.py", line 1548, in _send_frame
raise exceptions.ConnectionClosed
aio_pika.pika.exceptions.ConnectionClosed
except:
print("connection problems..")
这将赶上像服务异常KeyboardInterrupt，SystemExit。你应该永远不会做这样的事情，如果你不打算再加注例外。至少您应该写：
except Exception:
print("connection problems..")
但是在asyncio上面的代码片段的上下文中，将破坏取消机制。为了避免它作为解释在这里你应该写：
try:
await operation
except asyncio.CancelledError:
raise
except Exception:
log.log('an error has occurred')
同样重要的是要理解不仅应该打开连接，而且应该关闭连接（无论打开和关闭之间发生了什么）。为了实现这一目标，人们通常使用上下文管理器（以及asyncio异步上下文管理器）。
aio_pika似乎也不例外。如示例所示，async with在处理连接时应使用：
connection = await aio_pika.connect_robust(
"amqp://guest:[email protected]/", loop=loop
)
async with connection:
# ...
本文收集自互联网，转载请注明来源。
如有侵权，请联系[email protected] 删除。
我来说两句