“发射后不管” python async/await

问题描述：

有时需要执行一些非关键的异步操作，但我不想等待它完成。在 Tornado 的协程实现中，您只需省略yield关键字即可“触发并忘记”异步函数。

我一直在尝试弄清楚如何使用 Python 3.5 中发布的新async/await语法来实现“触发并忘记”。例如，简化的代码片段：

async def async_foo():
    print("Do some stuff asynchronously here...")

def bar():
    async_foo()  # fire and forget "async_foo()"

bar()

但实际情况是，它bar()从未执行，而是我们收到运行时警告：

RuntimeWarning: coroutine 'async_foo' was never awaited
  async_foo()  # fire and forget "async_foo()"

解决方案 1：

更新：

如果您使用的是 Python >= 3.7，则请到处替换asyncio.ensure_future它。asyncio.create_task这是生成任务的一种更新、更好的方法。

asyncio.Task 改为“发射后不管”

根据 python 文档，asyncio.Task可以启动一些协程以“在后台”执行。 创建的任务asyncio.ensure_future不会阻止执行（因此函数将立即返回！）。 这看起来像是一种“触发并忘记”的方式，正如您所要求的那样。

import asyncio


async def async_foo():
    print("async_foo started")
    await asyncio.sleep(1)
    print("async_foo done")


async def main():
    asyncio.ensure_future(async_foo())  # fire and forget async_foo()

    # btw, you can also create tasks inside non-async funcs

    print('Do some actions 1')
    await asyncio.sleep(1)
    print('Do some actions 2')
    await asyncio.sleep(1)
    print('Do some actions 3')


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

输出：

Do some actions 1
async_foo started
Do some actions 2
async_foo done
Do some actions 3

如果事件循环完成后任务正在执行会怎么样？

请注意，asyncio 期望任务在事件循环完成时完成。因此，如果您将其更改main()为：

async def main():
    asyncio.ensure_future(async_foo())  # fire and forget

    print('Do some actions 1')
    await asyncio.sleep(0.1)
    print('Do some actions 2')

程序运行完成后你会收到以下警告：

Task was destroyed but it is pending!
task: <Task pending coro=<async_foo() running at [...]

为了防止这种情况，您可以在事件循环完成后等待所有待处理的任务：

async def main():
    asyncio.ensure_future(async_foo())  # fire and forget

    print('Do some actions 1')
    await asyncio.sleep(0.1)
    print('Do some actions 2')


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    
    # Let's also finish all running tasks:
    pending = asyncio.Task.all_tasks()
    loop.run_until_complete(asyncio.gather(*pending))

终止任务而不是等待它们

有时你不想等待任务完成（例如，某些任务可能会被创建为永远运行）。在这种情况下，你可以直接执行cancel()它们而不是等待它们：

import asyncio
from contextlib import suppress


async def echo_forever():
    while True:
        print("echo")
        await asyncio.sleep(1)


async def main():
    asyncio.ensure_future(echo_forever())  # fire and forget

    print('Do some actions 1')
    await asyncio.sleep(1)
    print('Do some actions 2')
    await asyncio.sleep(1)
    print('Do some actions 3')


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

    # Let's also cancel all running tasks:
    pending = asyncio.Task.all_tasks()
    for task in pending:
        task.cancel()
        # Now we should await task to execute it's cancellation.
        # Cancelled task raises asyncio.CancelledError that we can suppress:
        with suppress(asyncio.CancelledError):
            loop.run_until_complete(task)

输出：

Do some actions 1
echo
Do some actions 2
echo
Do some actions 3
echo

解决方案 2：

输出：

>>> Hello
>>> foo() started
>>> I didn't wait for foo()
>>> foo() completed

这是一个简单的装饰函数，它将执行推送到后台，并将控制线移动到代码的下一行。

主要优点是，你不必将函数声明为await

import asyncio
import time

def fire_and_forget(f):
    def wrapped(*args, **kwargs):
        return asyncio.get_event_loop().run_in_executor(None, f, *args, *kwargs)

    return wrapped

@fire_and_forget
def foo():
    print("foo() started")
    time.sleep(1)
    print("foo() completed")

print("Hello")
foo()
print("I didn't wait for foo()")

注意：检查我的其他答案thread，它使用不带 的普通方法执行相同操作asyncio。

解决方案 3：

这不是完全异步执行，但也许run_in_executor()适合您。

def fire_and_forget(task, *args, **kwargs):
    loop = asyncio.get_event_loop()
    if callable(task):
        return loop.run_in_executor(None, task, *args, **kwargs)
    else:    
        raise TypeError('Task must be a callable')

def foo():
    #asynchronous stuff here


fire_and_forget(foo)

解决方案 4：

如果您出于某种原因无法使用asyncio，那么这里是使用普通线程的实现。请查看我的其他答案以及 Sergey 的答案。

import threading, time

def fire_and_forget(f):
    def wrapped():
        threading.Thread(target=f).start()

    return wrapped

@fire_and_forget
def foo():
    print("foo() started")
    time.sleep(1)
    print("foo() completed")

print("Hello")
foo()
print("I didn't wait for foo()")

生产

>>> Hello
>>> foo() started
>>> I didn't wait for foo()
>>> foo() completed

解决方案 5：

存在未指定的终止问题，因为“发射后不管”没有说明活动必须何时完成（以及最终在程序终止时是否挂起或终止它们）。解决方案是使用上下文管理器。Python 3.11 现在将其作为TaskGroup。

较早的替代方案是aiowire包。它的上下文管理器具有超时选项，并使用“蹦床”设计，允许异步函数返回异步函数。这既避免了后台线程，也避免了无限的异步调用链。

解决方案 6：

def fire_and_forget(f):
    def wrapped(*args, **kwargs):
        threading.Thread(target=functools.partial(f, *args, **kwargs)).start()

    return wrapped

是上面的更好版本——不使用 asyncio