Python异步loop.run_forever()

Question

我试着做一个交易机器人，而且大多数事情都很好。但是每次互联网连接消失一段时间，代码就会失败。我使用异步run_forever()函数，我认为代码应该永远运行，直到它被停止为止，但是它不能工作。

这是我的代码：

import json


async def listen():
    url = "wss://phemex.com/ws"
    async with websockets.connect(url) as ws:
        sub_msg = json.dumps({
            "id": sequence,
            "method": "kline.subscribe",
            "params": [symbol, kline_interval],
        })
        await ws.send(sub_msg)
        while True:
            msg = await ws.recv()
            msg = json.loads(msg)["kline"]

我把这个循环叫做：

loop = asyncio.get_event_loop()
try: 
    loop.create_task(listen())
    asyncio.ensure_future(listen())
    loop.run_forever()
except KeyboardInterrupt:
    pass
finally:
    print("Closing Loop")
    loop.close()

一旦连接丢失，就会出现以下错误：

Task exception was never retrieved
future: <Task finished coro=<listen() done, defined at c:\Users\danis\Desktop\AWS\Dateien\Crypto_Bot_Telegram_PSAR_PHEMEX_inkl_websocket_reconnect.py:351> exception=ConnectionClosedError('code = 1006 (connection closed abnormally [internal]), no reason',)>
Traceback (most recent call last):
  File "C:\Users\danis\.conda\envs\python36\lib\site-packages\websockets\legacy\protocol.py", line 750, in transfer_data
    message = await self.read_message()
  File "C:\Users\danis\.conda\envs\python36\lib\site-packages\websockets\legacy\protocol.py", line 819, in read_message
    frame = await self.read_data_frame(max_size=self.max_size)
  File "C:\Users\danis\.conda\envs\python36\lib\site-packages\websockets\legacy\protocol.py", line 895, in read_data_frame
    frame = await self.read_frame(max_size)
  File "C:\Users\danis\.conda\envs\python36\lib\site-packages\websockets\legacy\protocol.py", line 975, in read_frame
    extensions=self.extensions,
  File "C:\Users\danis\.conda\envs\python36\lib\site-packages\websockets\legacy\framing.py", line 55, in read
    data = await reader(2)
  File "C:\Users\danis\.conda\envs\python36\lib\asyncio\streams.py", line 674, in readexactly
    yield from self._wait_for_data('readexactly')
  File "C:\Users\danis\.conda\envs\python36\lib\asyncio\streams.py", line 464, in _wait_for_data
    yield from self._waiter
  File "C:\Users\danis\.conda\envs\python36\lib\asyncio\selector_events.py", line 714, in _read_ready
    data = self._sock.recv(self.max_size)
ConnectionResetError: [WinError 10054] Eine vorhandene Verbindung wurde vom Remotehost geschlossen

我怎样才能永远运行这段代码？

Answer 1

在asyncio中，事件循环负责运行异步任务，但它不能处理它们可能抛出的错误。有不同的方法来运行循环，您可以运行它来执行特定的任务，或者永远运行它，所以它一直在等待新任务的运行。

在您的示例中，任务listen正在抛出一个未捕获的异常(ConnectionResetError)，事件循环会使您对此与回溯产生怀疑，但它将继续运行，可能与其他任务一起运行(这意味着永远运行)。您刚刚收到通知，就像协同工作中发生的错误一样，任务停止运行。

解决方案：处理库中回溯的错误，您必须确保它将永远运行，事件循环不会这样做。

async def listen():
    url = "wss://phemex.com/ws"
    while True:
        try:
            async with websockets.connect(url) as ws:
                sub_msg = "{\"id\":" + str(
                    sequence) + ", \"method\": \"kline.subscribe\", \"params\":[\"" + symbol + "\"," 
                    + str(kline_interval) + "]}"
                await ws.send(sub_msg)
                while True:
                    msg = await ws.recv()
                    msg = json.loads(msg)["kline"]
        except ConnectionResetError:
            print("ConnectionResetError, reconnecting...")

Answer 2

您的循环将永远运行，但它无法阻止程序在未处理的异常上崩溃。在您的示例中，您尝试从可能失去连接的客户端获取recv数据。你可以自己抓到这个例外！

while True:
    async with websockets.connect(url) as ws:
        sub_msg = "{\"id\":" + str(
            sequence) + ", \"method\": \"kline.subscribe\", \"params\":[\"" + symbol + "\"," 
            + str(kline_interval) + "]}"
        await ws.send(sub_msg)

        while True:
            try: 
                msg = await ws.recv()
            except ConnectionResetError:
                break
            msg = json.loads(msg)["kline"]