在 Python文档中，它指出：

应用程序开发人员通常应该使用高级 asyncio 函数，例如 asyncio.run()，并且很少需要引用循环对象或调用其方法。 本节主要面向需要更好地控制事件循环行为的低级代码、库和框架的作者。

当同时使用async和 a 时threadpoolexecutor，如示例代码中所示（来自文档）：

import asyncio
import concurrent.futures

def blocking_io():
    # File operations (such as logging) can block the
    # event loop: run them in a thread pool.
    with open('/dev/urandom', 'rb') as f:
        return f.read(100)

async def main():
    loop = asyncio.get_running_loop()

    # 2. Run in a custom thread pool:
    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(
            pool, blocking_io)
        print('custom thread pool', result)

asyncio.run(main()) 

1. 我需要打电话loop.close()，还是会asyncio.run() …

我想知道当我们await在异步 Python 代码中使用协程时究竟会发生什么，例如：

await send_message(string)

(1)send_message加入事件循环，调用协程放弃对事件循环的控制，或

(2) 我们直接跳进去 send_message

我读到的大多数解释都指向 (1)，因为它们将调用协程描述为exiting。但我自己的实验表明（2）是这种情况：我试图在调用者之后但在被调用者之前运行一个协程，但无法实现这一点。

我试图理解 asyncio 模块并花了大约一小时使用 run_coroutine_threadsafe 函数，我什至来到了工作示例，它按预期工作，但有一些限制。

首先，我不明白应该如何在主（任何其他）线程中正确调用 asyncio 循环，在示例中我调用它并run_until_complete给它一个协程以使其忙于某些事情，直到另一个线程不会给它一个协程。我还有什么其他选择？

在现实生活中，什么情况下我必须混合使用异步和线程（在 Python 中）？据我了解，asyncio 应该取代 Python 中的线程（由于 GIL 不是 IO 操作），如果我错了，请不要生气并分享您的建议。

Python版本为3.7/3.8

import asyncio
import threading
import time


async def coro_func():
    return await asyncio.sleep(3, 42)


def another_thread(_loop):
    coro = coro_func()  # is local thread coroutine which we would like to run in another thread

    # _loop is a loop which was created in another thread

    future = asyncio.run_coroutine_threadsafe(coro, _loop)
    print(f"{threading.current_thread().name}: {future.result()}")
    time.sleep(15)
    print(f"{threading.current_thread().name} is Finished")


if __name__ == '__main__':
    loop = …

假设我有一个像这样的异步生成器：

async def event_publisher(connection, queue):
    while True:
        if not await connection.is_disconnected():
            event = await queue.get()
            yield event
        else:
            return

我是这样消费的：

published_events = event_publisher(connection, queue)
async for event in published_events:
    # do event processing here

它工作得很好，但是当连接断开并且没有发布新事件时，async for它将永远等待，所以理想情况下，我想像这样强行关闭生成器：

if connection.is_disconnected():
    await published_events.aclose()

但我收到以下错误：

RuntimeError: aclose(): asynchronous generator is already running

有没有办法停止处理已经运行的发电机？

我有一个在 fastapi 和 aiofiles 上运行的异步代码，我正在尝试从 .json 文件加载和保存我的信息，但是每次我关闭程序时，它只保存字典的键并显示“ASGI 'lifespan'协议似乎不受支持”按摩

这是我的开启/关闭部分：

@app.on_event("startup")
async def startup_event():
    global beers
    try:
        async with aiofiles.open("data.json", mode='r+', json=True) as file:
            beers = await file.read()
    except:
        beers = {}


@app.on_event("shutdown")
async def on_exit_app():
    async with aiofiles.open("data.json", "w+") as outfile:
        await outfile.write(beers)

任何想法问题出在哪里？

鉴于 celery 使用这些选项运行：

celery -A openwisp2 worker -l info --pool=gevent --concurrency=15 -Ofair

鉴于来自 openwisp-monitoring 的这个celery 任务：

@shared_task
def perform_check(uuid):
    """
    Retrieves check according to the passed UUID
    and calls ``check.perform_check()``
    """
    try:
        check = get_check_model().objects.get(pk=uuid)
    except ObjectDoesNotExist:
        logger.warning(f'The check with uuid {uuid} has been deleted')
        return
    result = check.perform_check()
    if settings.DEBUG:  # pragma: nocover
        print(json.dumps(result, indent=4, sort_keys=True))

大部分时间任务都可以工作，但有时（通常是突发），会生成以下异常：

SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.

完整的堆栈跟踪：

SynchronousOnlyOperation: You …

如果我不使用await调用异步函数，我将返回一个协程。那么，协程会发生什么情况呢？我必须手动执行协程吗？或者这个协程将继续在后台运行？

使用await

async def work():
    result = await stuff()

没有await

async def work():
     result = stuff()