试图了解 Tasks 如何处理阻塞调用

Question

我最近开始研究 asyncio 库，目的是用异步替换基于大线程的应用程序。

阅读asyncio documentantion 我偶然发现了一个例子 create_task 正在使用中。因为我现在被python 3.6困住了，所以我改变了create_task调用 ensure_future，产生当前代码：

# Python 3.6
import asyncio
import time


async def say_after(delay, what):
    print(f"start {what}")  # Added for better vizualization of what is happening
    await asyncio.sleep(delay)
    print(what)


async def main():
    task1 = asyncio.ensure_future(
        say_after(1, 'hello'))

    task2 = asyncio.ensure_future(
        say_after(2, 'world'))

    print(f"started at {time.strftime('%X')}")

    # Wait until both tasks are completed (should take
    # around 2 seconds.)
    await task1
    await task2

    print(f"finished at {time.strftime('%X')}")


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

还有输出：

started at 15:23:11
start hello
start world
hello
world
finished at 15:23:13

据我了解，事件循环：

• 首先启动任务task1；
• 点击asyncio.sleep 后，它会将上下文更改为第二个task2；和
• 当第一次睡眠结束时，它会变为task1，同样的事情发生在task2 的睡眠调用结束时。

综上所述，我的应用程序的要求之一是我们有一些需要转换为协程的阻塞调用。 我为此目标创建了这个模拟代码来测试run_in_executor 函数：

# Python 3.6
import asyncio
import time

from concurrent.futures import ThreadPoolExecutor


def normal_operations():
    print("start blocking")
    time.sleep(1)
    print("ended blocking")


async def async_operation():
    print("start non blocking")
    await asyncio.sleep(2)
    print("ended non blocking")


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

    print(f"started at {time.strftime('%X')}")

    with ThreadPoolExecutor() as pool:
        task1 = asyncio.ensure_future(
            loop.run_in_executor(pool, normal_operations)
        )

    task2 = asyncio.ensure_future(
        async_operation()
    )

    await task1
    await task2

    print(f"finished at {time.strftime('%X')}")


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

我希望输出类似于第一个示例，但是当我运行此代码时，输​​出是：

started at 15:28:06
start blocking
ended blocking
start non blocking
ended non blocking
finished at 15:28:09

这两个函数是按顺序运行的，不像第一个示例，其中 start print 调用是在另一个之前调用的。

我不确定我做错了什么，我的猜测是run_in_executor 函数并没有真正创建异步调用，或者我只是执行错误，我不知道。

Answer 1

好的，我认为是错误的。

我正在等待ThreadPoolExecutor 之外的非阻塞操作，而在__exit__ dunder 中，shutdown 函数正在使用wait=True 参数调用，所以基本上执行程序阻塞了我的代码。

固定代码：

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

    print(f"started at {time.strftime('%X')}")

    pool = ThreadPoolExecutor()

    task1 = asyncio.ensure_future(loop.run_in_executor(pool, normal_operations))
    task2 = asyncio.ensure_future(async_operation())

    await task1
    await task2

    print(f"finished at {time.strftime('%X')}")

    pool.shutdown(wait=True)

预期输出：

started at 16:20:48
start non blocking
start blocking
ended blocking
ended non blocking
finished at 16:20:50