如何使用asyncio计划​​和取消任务

Question

我正在编写客户端 - 服务器应用程序.连接时,客户端每秒向服务器发送一个"心跳"信号.在服务器端,我需要一种机制,我可以在其中添加异步执行的任务(或协程或其他).此外,当我停止发送"心跳"信号时,我想从客户端取消任务.

换句话说,当服务器启动任务时,它具有一种超时或ttl,例如3秒.当服务器收到"心跳"信号时,它会将计时器重置另外3秒,直到任务完成或客户端断开连接(停止发送信号).

以下是从pymotw.com上的asyncio教程取消任务的示例.但是这里的任务在event_loop开始之前被取消了,这对我来说并不合适.

import asyncio

async def task_func():
    print('in task_func')
    return 'the result'


event_loop = asyncio.get_event_loop()
try:
    print('creating task')
    task = event_loop.create_task(task_func())

    print('canceling task')
    task.cancel()

    print('entering event loop')
    event_loop.run_until_complete(task)
    print('task: {!r}'.format(task))
except asyncio.CancelledError:
    print('caught error from cancelled task')
else:
    print('task result: {!r}'.format(task.result()))
finally:
    event_loop.close()

Answer 1

您可以使用asyncio Task包装器通过该ensure_future()方法执行任务.

ensure_future将自动将您的协同程序Task包装在一个包装器中并将其附加到您的事件循环中.在Task随后的包装也将确保协程从"怪人,在" await以await声明(或直至完成协程).

换句话说,只需将常规协程传递ensure_future给并将结果Task对象赋值给变量.然后,您可以Task.cancel()在需要停止时拨打电话.

import asyncio

async def task_func():
    print('in task_func')
    # if the task needs to run for a while you'll need an await statement
    # to provide a pause point so that other coroutines can run in the mean time
    await some_db_or_long_running_background_coroutine()
    # or if this is a once-off thing, then return the result,
    # but then you don't really need a Task wrapper...
    # return 'the result'

async def my_app():
    my_task = None
    while True:
        await asyncio.sleep(0)

        # listen for trigger / heartbeat
        if heartbeat and not my_task:
            my_task = asyncio.ensure_future(task_func())

        # also listen for termination of hearbeat / connection
        elif not heartbeat and my_task:
            if not my_task.cancelled():
                my_task.cancel()
            else:
                my_task = None

run_app = asyncio.ensure_future(my_app())
event_loop = asyncio.get_event_loop()
event_loop.run_forever()

请注意,任务适用于需要在后台继续工作而不会中断主流的长时间运行的任务.如果您只需要一个快速的一次性方法,那么只需直接调用该函数即可.