深入Python中的asyncio库——线程同步

前面的代码都是异步的，就如sleep，需要用asyncio.sleep而不是阻塞的time.sleep，如果有同步逻辑，怎么利用asyncio实现并发呢？答案是用run_in_executor。在一开始我说过开发者创建 Future 对象情况很少，主要是用run_in_executor，就是让同步函数在一个执行器( executor)里面运行。

同步代码

def a():
    time.sleep(1)
    return 'A'
async def b():
    await asyncio.sleep(1)
    return 'B'
def show_perf(func):
    print('*' * 20)
    start = time.perf_counter()
    asyncio.run(func())
    print(f'{func.__name__} Cost: {time.perf_counter() - start}')
async def c1():
    loop = asyncio.get_running_loop()
    await asyncio.gather(
        loop.run_in_executor(None, a),
        b()
    )
In : show_perf(c1)
********************
c1 Cost: 1.0027242230000866

可以看出，同步函数逻辑可以用run_into_executor转换成协程，实现并发。这里注意细节，即函数a是普通函数，不能写成协程。下面的定义是错误的，不能实现并发：

async def a():
    time.sleep(1)
    return 'A'

因为a中没有异步代码，所以不要用async def来定义。这个逻辑需要用loop.run_in_executor封装到协程中：

async def c():
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, a)

大家理解了吧？

loop.run_in_executor(None, a) 中的第一个参数是传递concurrent.futures.Executor 的一个实例。传递 None 将选择默认执行器：

In : loop._default_executor
Out: <concurrent.futures.thread.ThreadPoolExecutor at 0x112b60e80>

当然我们也可以使用进程池，这次改普通文件读写的例子，使用：

async def c3():
    loop = asyncio.get_running_loop()
    with concurrent.futures.ProcessPoolExecutor() as e:
        print(await asyncio.gather(
            loop.run_in_executor(e, a),
            b()
        ))
In : show_perf(c3)
********************
['A', 'B']
c3 Cost: 1.0218078890000015

下一节：