Python 异步协程 async/await/asyncio

你的名字 2023-07-18 12:48 3阅读 0赞

### 文章目录 ###

*  异步IO asyncio协程
 *  协程常见名字
 *  创建一个协程 Coroutine
 *  创建一个任务 Task
 *  获取返回结果 回调函数future/直接使用Task获取
 *  阻塞 await
 *  asyncio 实现并发
 *  协程嵌套
 *  协程的停止

# 异步IO asyncio协程 #

asyncio 是python3.4 引入的，内置的对异步IO的支持。

asyncio编程是一个消息循环，我们从asyncio中获取一个EventLoop引用

然后把协程放到EventLoop中执行，就实现了异步IO

# 协程常见名字 #

先了解以下名词,有助于理解程序编写

*  event\_loop 事件循环： 程序开启一个循环，我们把函数注册到loop里，满足条件就会执行
 *  coroutine 协程: async def 定义的函数，它不能直接执行,返回的是`协程对象`,它要注册到loop中，由loop去执行调用
 *  task 任务： 一个协程对象就是一个原生可以挂起的函数，任务则是对协程对象进一步封装，增加了任务的各种状态，（比如未执行，执行中，执行结束，返回结果等等？。）
    
    task 是 future的`子类`
 *  future : 代表将来执行或没有执行的任务的结果，它个task没有本质区别
 *  async/await : python3.5 用于定义协程的关键字

# 创建一个协程 Coroutine #

import asyncio,time
    
    # 记录开始时间
    now = lambda: time.time()
    start = now()
    
    # 01.定义一个协程
    async def do_work(num):
        print("num:",num)
    
    coroutine = do_work(1)
    
    # 02.创建事件循环
    loop = asyncio.get_event_loop()
    
    # 03.将协程对象放入事件循环中
    loop.run_until_complete(coroutine)
    
    # 记录结束时间
    print(now() - start)

# 创建一个任务 Task #

loop.run\_until\_complete方法传入协程，自动将协程装成任务task，它也可以直接接收task

**创建task** 使用以下方法都可以

asyncio.ensure\_future(coroutine)

loop.create\_task(coroutine)

task是future的子类。

isinstance(task,asyncio.Future) 返回Ture

import asyncio,time
    
    # 记录开始时间
    now = lambda: time.time()
    start = now()
    
    # 01.定义一个协程
    async def do_work(num):
        print("num:",num)
    
    coroutine = do_work(1)
    
    # 02.创建事件循环
    loop = asyncio.get_event_loop()
    
    # 03.创建任务Task
    task = asyncio.ensure_future(coroutine)
    # 或者下面的方式 创建task也可以
    # task = loop.create_task(coroutine)
    
    # 04.将任务注册到事件循环中
    loop.run_until_complete(task)
    
    # 记录结束时间
    print(now() - start)

# 获取返回结果 回调函数future/直接使用Task获取 #

import asyncio,time
    
    # 记录开始时间
    now = lambda: time.time()
    start = now()
    
    # 01.定义一个协程
    async def do_work(num):
        print("num:",num)
        return "num is {}".format(num)
    
    coroutine = do_work(3)
    
    
    # 02.定义回调函数
    def callback(future):
        print('result is',future.result())
    
    # 03.创建事件循环
    loop = asyncio.get_event_loop()
    
    # 04.创建任务Task
    task = loop.create_task(coroutine)
    
    # 05.给task任务绑定回调函数
    task.add_done_callback(callback)
    
    # 06.将任务注册到事件循环中
    loop.run_until_complete(task)
    
    # 记录结束时间
    print(now() - start)

不绑定回调，直接使用Task的结果

`task.result()`，必须任务执行完毕后才有这个，不然会报错

# 阻塞 await #

import asyncio,time
    
    # 记录开始时间
    now = lambda: time.time()
    start = now()
    
    # 01.定义一个协程
    async def do_work(num):
        print("num:",num)
        await asyncio.sleep(num)
        return "sleep is {}".format(num)
    
    coroutine = do_work(3)
    
    
    # 02.创建事件循环
    loop = asyncio.get_event_loop()
    
    # 03.创建任务Task
    task = loop.create_task(coroutine)
    print(task.result())
    
    # 04.将任务注册到事件循环中
    loop.run_until_complete(task)
    
    # 记录结束时间
    print(now() - start)
    print(task.result())

# asyncio 实现并发 #

import asyncio,time
    
    # 记录开始时间
    now = lambda: time.time()
    start = now()
    
    # 01.定义一个协程
    async def do_work(num):
        print("num:",num)
        await asyncio.sleep(num)
        return "sleep is {}".format(num)
    
    coroutine1 = do_work(1)
    coroutine2 = do_work(2)
    coroutine3 = do_work(3)
    
    
    
    # 02.创建任务Task
    tasks =[
        asyncio.ensure_future(coroutine1),
        asyncio.ensure_future(coroutine2),
        asyncio.ensure_future(coroutine3),
        
    ]
    
    # 02.创建事件循环
    loop = asyncio.get_event_loop()
    
    # 04.将任务注册到事件循环中
    loop.run_until_complete(asyncio.wait(tasks))
    
    # 记录结束时间
    print(now() - start)
    for task in tasks:
        print(task.result())

# 协程嵌套 #

import time
    import asyncio
    
    
    async def do_work(x):
        print("do...work.....")
        await asyncio.sleep(x)
        return 'Done result'
    
    
    async def main():
        # 创建多个协程对象
        coroutine1 = do_work(1)
        coroutine2 = do_work(2)
        coroutine3 = do_work(4)
        
        tasks = [
            asyncio.ensure_future(coroutine1),
            asyncio.ensure_future(coroutine2),
            asyncio.ensure_future(coroutine3)
        ]
        
        # --------------------------------------------
        # 获取返回结果的方式 1
        dones,pendings = await asyncio.wait(tasks)
        # dones 是任务
        for task in dones:
            print(task.result())
        # --------------------------------------------
        
    # 将协程main 放入到 loop中
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

# 获取返回结果的方式2
    results =  await asyncio.gather(*tasks)
    for result in results():
        print(result)

# 获取方式3,不在main中获取，而是在外面获取
    async def main:
        # .....
        return await asyncio.gather(*tasks)
    
    results = loop.run_until_complete(main())
    for result in results():
        print(result)

# 获取方式4,不在main中获取，而是在外面获取
    async def main:
        # .....
        return await asyncio.wait(tasks)
    
    dones,pendings = loop.run_until_complete(main())
    for task in dones:
        print(task.result())

# 获取方式5,还是在main中
    
    for task in asyncio.as_complete(tasks):
        result = await task
        print(result)

# 协程的停止 #

future 中的几种状态

*  Pending 创建future的时候，task为 pending
 *  Running 事件循环执行的时候，task是running
 *  Done 执行完毕 task状态是done
 *  Cancelled 任务取消

import time
    import asyncio
    
    
    async def do_work(x):
        print("do...work.....")
        await asyncio.sleep(x)
        return 'Done result'
    
    
    coroutine1 = do_work(1)
    coroutine2 = do_work(2)
    coroutine3 = do_work(4)
    
    
    tasks = [
            asyncio.ensure_future(coroutine1),
            asyncio.ensure_future(coroutine2),
            asyncio.ensure_future(coroutine3),
    ]
    
    
    start = time.time()
    loop = asyncio.get_event_loop()
    
    try:
        loop.run_until_complete(asyncio.wait(tasks))
    except KeyboardInterrupt as e:
        for task in asyncio.Task.all_tasks():
            print(task.cancel())
        loop.stop()
        loop.run_forever()
    finally:
        loop.close()
    
    print(time.time() - start)