Python库asyncio正确处理blocking functions

发表于 | 更新于 | 分类于

问题

问题摘要

当我在写一段需要高性能同时运行的代码时出现了一个奇怪的现象,及当我加上一个未被调用的函数时执行时间发生了变化,在Pycharm中变慢了,在命令行中变快了

示例代码放在 这里 ,其中包含了在google colab上的运行时间

问题代码

初始代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
def main2():
    async def f1():
        await asyncio.sleep(0.1)

    async def f3(im, i):
        for _ in range(60):
            g = Image.fromarray(im).convert("L")
            g.save("./{}.png".format(str(i)))

    im = np.random.randint(0, 256, (240, 256, 3)).astype("uint8")

    loop = asyncio.get_event_loop()
    s = time.time()
    for i in range(10):
        result = loop.run_until_complete(
            asyncio.gather(
                f1(),
                f3(im, i)
            )
        )
    print(time.time() - s)

改动后代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def main():
    async def f1():
        await asyncio.sleep(0.1)
    
    # 这个函数是新加的,但是没有在任何地方调用,但速度却改变了
    async def f2(im, i):
        for j in range(60):
            Image.fromarray(im).convert("L").save("./{}.png".format(str(i)))

    async def f3(im, i):
        for _ in range(60):
            g = Image.fromarray(im).convert("L")
            g.save("./{}.png".format(str(i)))

    im = np.random.randint(0, 256, (240, 256, 3)).astype("uint8")

    loop = asyncio.get_event_loop()
    s = time.time()
    for i in range(10):
        result = loop.run_until_complete(
            asyncio.gather(
                f1(),
                f3(im, i)
            )
        )
    print(time.time() - s)

解决

首先先展示一下正确的代码写法

修正代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def main3():
    def f1():
        time.sleep(0.1)
        return 1

    def f3(im, i):
        for _ in range(60):
            g = Image.fromarray(im).convert("L")
            g.save("./{}.png".format(str(i)))
        return 3

    im = np.random.randint(0, 256, (240, 256, 3)).astype("uint8")

    loop = asyncio.get_event_loop()
    executor = futures.ThreadPoolExecutor(max_workers=10)

    s = time.time()

    for i in range(10):
        coroutine_list = [loop.run_in_executor(executor, f3, *(im,i)),
                  loop.run_in_executor(executor, f1, )]
        task = asyncio.gather(*coroutine_list)
        task = asyncio.ensure_future(task)
        loop.run_until_complete(task)
        # print(task.result())

    print(time.time() - s)

原因

f3函数并不支持协程调用,或者说与asyncio不能很好的兼容工作,使得代码的行为不可控制.

查看官方网站后发现一个专门用来处理这种函数的方法 run_in_executor , 当使用这个后,f1,f3是同步运行的

参考资料

官方资料:Handle blocking functions correctly

一个解释syncio的很好的博客:Python黑魔法 — 异步IO（ asyncio） 协程