Python asyncio 重启协程

Question

我是一名新的 Python 程序员，他正在尝试编写一个“机器人”来为我自己在 betfair 上进行交易。 (雄心勃勃!!!!)

我出现的问题是——我有运行异步事件循环的基础知识，但我注意到如果其中一个协程在其过程中失败(例如 API 调用失败或 mongodb 读取)，那么异步事件循环只是继续运行但忽略了一个失败的协程

我的问题是，我如何才能自动重启那个协程或处理错误以停止整个异步循环，但目前一切运行似乎都没有意识到某些事情不对劲并且其中一部分失败了这一事实。在我的例子中，在数据库读取不成功之后，循环永远不会返回到“rungetcompetitionids”函数，即使它处于 while true 循环中，它也永远不会再次返回到该函数usergui 还没有功能，只是用来尝试 asyncio

谢谢克莱夫

import sys
import datetime
from login import sessiontoken as gst
from mongoenginesetups.setupmongo import global_init as initdatabase
from asyncgetcompetitionids import competition_id_pass as gci
from asyncgetcompetitionids import create_comp_id_table_list as ccid
import asyncio
import PySimpleGUI as sg

sg.change_look_and_feel('DarkAmber')

layout = [
    [sg.Text('Password'), sg.InputText(password_char='*', key='password')],
    [sg.Text('', key='status')],
    [sg.Button('Submit'), sg.Button('Cancel')]
]
window = sg.Window('Betfair', layout)


def initialisethedatabase():
    initdatabase('xxxx', 'xxxx', xxxx, 'themongo1', True)


async def runsessiontoken():
    nextlogontime = datetime.datetime.now()
    while True:
        returned_login_time = gst(nextlogontime)
        nextlogontime = returned_login_time
        await asyncio.sleep(15)


async def rungetcompetitionids(compid_from_compid_table_list):
    nextcompidtime = datetime.datetime.now()
    while True:
        returned_time , returned_list = gci(nextcompidtime, compid_from_compid_table_list)
        nextcompidtime = returned_time
        compid_from_compid_table_list = returned_list
        await asyncio.sleep(10)


async def userinterface():
    while True:
        event, value = window.read(timeout=1)
        if event in (None, 'Cancel'):
            sys.exit()
        if event != "__TIMEOUT__":
            print(f"{event} {value}")
        await asyncio.sleep(0.0001)


async def wait_list():
    await asyncio.wait([runsessiontoken(), 
                       rungetcompetitionids(compid_from_compid_table_list), 
                       userinterface()
                      ])


initialisethedatabase()
compid_from_compid_table_list = ccid()
print(compid_from_compid_table_list)
nextcompidtime = datetime.datetime.now()
print(nextcompidtime)
loop = asyncio.get_event_loop()
loop.run_until_complete(wait_list())
loop.close()

Answer 1

一个简单的解决方案是使用包装器函数(或“主管”)捕获 Exception，然后盲目地重试该函数。更优雅的解决方案包括打印出异常和堆栈跟踪以用于诊断目的，以及查询应用程序状态以查看尝试和继续是否有意义。例如，如果必发告诉您您的帐户未经授权，那么继续操作就没有意义了。如果这是一个一般的网络错误，那么立即重新绑定(bind)可能是值得的。如果主管注意到它在短时间内重新启动了很多次，您可能还想停止重试。

例如。

import asyncio
import traceback
import functools
from collections import deque
from time import monotonic

MAX_INTERVAL = 30
RETRY_HISTORY = 3
# That is, stop after the 3rd failure in a 30 second moving window

def supervise(func, name=None, retry_history=RETRY_HISTORY, max_interval=MAX_INTERVAL):
    """Simple wrapper function that automatically tries to name tasks"""
    if name is None:
        if hasattr(func, '__name__'): # raw func
            name = func.__name__
        elif hasattr(func, 'func'): # partial
            name = func.func.__name__
    return asyncio.create_task(supervisor(func, retry_history, max_interval), name=name)


async def supervisor(func, retry_history=RETRY_HISTORY, max_interval=MAX_INTERVAL):
    """Takes a noargs function that creates a coroutine, and repeatedly tries
        to run it. It stops is if it thinks the coroutine is failing too often or
        too fast.
    """
    start_times = deque([float('-inf')], maxlen=retry_history)
    while True:
        start_times.append(monotonic())
        try:
            return await func()
        except Exception:
            if min(start_times) > monotonic() - max_interval:
                print(
                    f'Failure in task {asyncio.current_task().get_name()!r}.'
                    ' Is it in a restart loop?'
                )
                # we tried our best, this coroutine really isn't working.
                # We should try to shutdown gracefully by setting a global flag
                # that other coroutines should periodically check and stop if they
                # see that it is set. However, here we just reraise the exception.
                raise
            else:
                print(func.__name__, 'failed, will retry. Failed because:')
                traceback.print_exc()


async def a():
    await asyncio.sleep(2)
    raise ValueError


async def b(greeting):
    for i in range(15):
        print(greeting, i)
        await asyncio.sleep(0.5)


async def main_async():
    tasks = [
        supervise(a),
        # passing repeated argument to coroutine (or use lambda)
        supervise(functools.partial(b, 'hello'))
    ]
    await asyncio.wait(
        tasks,
        # Only stop when all coroutines have completed
        # -- this allows for a graceful shutdown
        # Alternatively use FIRST_EXCEPTION to stop immediately 
        return_when=asyncio.ALL_COMPLETED,
    )
    return tasks


def main():
    # we run outside of the event loop, so we can carry out a post-mortem
    # without needing the event loop to be running.
    done = asyncio.run(main_async())
    for task in done:
        if task.cancelled():
            print(task, 'was cancelled')
        elif task.exception():
            print(task, 'failed with:')
            # we use a try/except here to reconstruct the traceback for logging purposes
            try:
                task.result()
            except:
                # we can use a bare-except as we are not trying to block
                # the exception -- just record all that may have happened.
                traceback.print_exc()

main()

这将导致如下输出:

hello 0hello 1hello 2hello 3a failed, will retry. Failed because:Traceback (most recent call last):  File "C:\Users\User\Documents\python\src\main.py", line 30, in supervisor    return await func()  File "C:\Users\User\Documents\python\src\main.py", line 49, in a    raise ValueErrorValueErrorhello 4hello 5hello 6hello 7a failed, will retry. Failed because:Traceback (most recent call last):  File "C:\Users\User\Documents\python\src\main.py", line 30, in supervisor    return await func()  File "C:\Users\User\Documents\python\src\main.py", line 49, in a    raise ValueErrorValueErrorhello 8hello 9hello 10hello 11Failure in task 'a'. Is it in a restart loop?hello 12hello 13hello 14 exception=ValueError()> failed with:Traceback (most recent call last):  File "C:\Users\User\Documents\python\src\main.py", line 84, in main    task.result()  File "C:\Users\User\Documents\python\src\main.py", line 30, in supervisor    return await func()  File "C:\Users\User\Documents\python\src\main.py", line 49, in a    raise ValueErrorValueError