Asyncio 在单独的线程中“触发并忘记”任务

Question

我有一个长时间运行的同步 Python 程序，我希望每秒运行大约 10 个“即发即弃”的任务。这些任务命中远程 API，不需要返回任何值。我尝试了this answer，但它需要太多的 CPU/内存来生成和维护所有单独的线程，所以我一直在研究 asyncio。

This answer 很好地解释了如何使用 asyncio 运行“一劳永逸”。但是，它需要使用run_until_complete()，它会等待所有异步任务完成。我的程序正在使用同步 Python，所以这对我不起作用。理想情况下，代码应该像这样简单，log_remote 不会阻塞循环：

while True:
    latest_state, metrics = expensive_function(latest_state)
    log_remote(metrics) # <-- this should be run as "fire and forget"

我使用的是 Python 3.7。如何在另一个线程上使用 asyncio 轻松运行它？

Answer 1

您可以在单个后台线程中启动单个事件循环，并将其用于您的所有即发即弃任务。例如：

import asyncio, threading

_loop = None

def fire_and_forget(coro):
    global _loop
    if _loop is None:
        _loop = asyncio.new_event_loop()
        threading.Thread(target=_loop.run_forever, daemon=True).start()
    _loop.call_soon_threadsafe(asyncio.create_task, coro)

有了这个，你就可以在通过调用async def创建的协程对象上调用fire_and_forget：

# fire_and_forget defined as above

import time

async def long_task(msg):
    print(msg)
    await asyncio.sleep(1)
    print('done', msg)

fire_and_forget(long_task('foo'))
fire_and_forget(long_task('bar'))
print('continuing with something else...')
time.sleep(3)

请注意，log_remote 需要使用 asyncio、aiohttp 而不是请求等实际编写为 async def。