zoukankan      html  css  js  c++  java
  • 异步IO框架:asyncio 中篇

    上一节我们首先介绍了,如何创建一个协程对象.
    主要有两种方法

    • 通过async关键字,
    • 通过@asyncio.coroutine 装饰函数。

    然后有了协程对象,就需要一个事件循环容器来运行我们的协程。其主要的步骤有如下几点:

    • 将协程对象转为task任务对象
    • 定义一个事件循环对象容器用来存放task
    • 将task任务扔进事件循环对象中并触发

    上一节,其实就只是讲了协程中的单任务,我们就来看下,协程中的多任务。

    协程中的并发

    协程的并发,和线程一样。举个例子来说,就好像 一个人同时吃三个馒头,咬了第一个馒头一口,就得等这口咽下去,才能去啃第其他两个馒头。就这样交替换着吃。

    asyncio实现并发,就需要多个协程来完成任务,每当有任务阻塞的时候就await,然后其他协程继续工作。

    第一步,当然是创建多个协程的列表。

    # 协程函数
    async def do_some_work(x):
        print('Waiting: ', x)
        await asyncio.sleep(x)
        return 'Done after {}s'.format(x)
    
    # 协程对象
    coroutine1 = do_some_work(1)
    coroutine2 = do_some_work(2)
    coroutine3 = do_some_work(4)
    
    # 将协程转成task,并组成list
    tasks = [
        asyncio.ensure_future(coroutine1),
        asyncio.ensure_future(coroutine2),
        asyncio.ensure_future(coroutine3)
    ]

    第二步,如何将这些协程注册到事件循环中呢。

    有两种方法,至于这两种方法什么区别,稍后会介绍。

    • 使用asyncio.wait()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))
    • 使用asyncio.gather()
    # 千万注意,这里的 「*」 不能省略
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.gather(*tasks))

    最后,return的结果,可以用task.result()查看。

    for task in tasks:
        print('Task ret: ', task.result())

    完整代码如下

    import asyncio
    
    # 协程函数
    async def do_some_work(x):
        print('Waiting: ', x)
        await asyncio.sleep(x)
        return 'Done after {}s'.format(x)
    
    # 协程对象
    coroutine1 = do_some_work(1)
    coroutine2 = do_some_work(2)
    coroutine3 = do_some_work(4)
    
    # 将协程转成task,并组成list
    tasks = [
        asyncio.ensure_future(coroutine1),
        asyncio.ensure_future(coroutine2),
        asyncio.ensure_future(coroutine3)
    ]
    
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))
    
    for task in tasks:
        print('Task ret: ', task.result())

    输出结果

    Waiting:  1
    Waiting: 2
    Waiting: 4
    Task ret: Done after 1s
    Task ret: Done after 2s
    Task ret: Done after 4s

    协程中的嵌套

    使用async可以定义协程,协程用于耗时的io操作,我们也可以封装更多的io操作过程,这样就实现了嵌套的协程,即一个协程中await了另外一个协程,如此连接起来。

    来看个例子。

    import asyncio
    
    # 用于内部的协程函数
    async def do_some_work(x):
        print('Waiting: ', x)
        await asyncio.sleep(x)
        return 'Done after {}s'.format(x)
    
    # 外部的协程函数
    async def main():
        # 创建三个协程对象
        coroutine1 = do_some_work(1)
        coroutine2 = do_some_work(2)
        coroutine3 = do_some_work(4)
    
        # 将协程转为task,并组成list
        tasks = [
            asyncio.ensure_future(coroutine1),
            asyncio.ensure_future(coroutine2),
            asyncio.ensure_future(coroutine3)
        ]
    
        # 【重点】:await 一个task列表(协程)
        # dones:表示已经完成的任务
        # pendings:表示未完成的任务
        dones, pendings = await asyncio.wait(tasks)
    
        for task in dones:
            print('Task ret: ', task.result())
    
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

    如果这边,使用的是asyncio.gather(),是这么用的

    # 注意这边返回结果,与await不一样
    
    results = await asyncio.gather(*tasks)
    for result in results:
        print('Task ret: ', result)

    输出还是一样的。

    Waiting:  1
    Waiting: 2
    Waiting: 4
    Task ret: Done after 1s
    Task ret: Done after 2s
    Task ret: Done after 4s

    仔细查看,可以发现这个例子完全是由 上面「协程中的并发」例子改编而来。结果完全一样。只是把创建协程对象,转换task任务,封装成在一个协程函数里而已。外部的协程,嵌套了一个内部的协程。

    其实你如果去看下asyncio.await()的源码的话,你会发现下面这种写法

    loop.run_until_complete(asyncio.wait(tasks))

    看似没有嵌套,实际上内部也是嵌套的。

    这里也把源码,贴出来,有兴趣可以看下,没兴趣,可以直接跳过。

    # 内部协程函数
    async def _wait(fs, timeout, return_when, loop):
        assert fs, 'Set of Futures is empty.'
        waiter = loop.create_future()
        timeout_handle = None
        if timeout is not None:
            timeout_handle = loop.call_later(timeout, _release_waiter, waiter)
        counter = len(fs)
    
        def _on_completion(f):
            nonlocal counter
            counter -= 1
            if (counter <= 0 or
                return_when == FIRST_COMPLETED or
                return_when == FIRST_EXCEPTION and (not f.cancelled() and
                                                    f.exception() is not None)):
                if timeout_handle is not None:
                    timeout_handle.cancel()
                if not waiter.done():
                    waiter.set_result(None)
    
        for f in fs:
            f.add_done_callback(_on_completion)
    
        try:
            await waiter
        finally:
            if timeout_handle is not None:
                timeout_handle.cancel()
    
        done, pending = set(), set()
        for f in fs:
            f.remove_done_callback(_on_completion)
            if f.done():
                done.add(f)
            else:
                pending.add(f)
        return done, pending
    
    # 外部协程函数
    async def wait(fs, *, loop=None, timeout=None, return_when=ALL_COMPLETED):
        if futures.isfuture(fs) or coroutines.iscoroutine(fs):
            raise TypeError(f"expect a list of futures, not {type(fs).__name__}")
        if not fs:
            raise ValueError('Set of coroutines/Futures is empty.')
        if return_when not in (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED):
            raise ValueError(f'Invalid return_when value: {return_when}')
    
        if loop is None:
            loop = events.get_event_loop()
    
        fs = {ensure_future(f, loop=loop) for f in set(fs)}
        # 【重点】:await一个内部协程
        return await _wait(fs, timeout, return_when, loop)

    协程中的状态

    还记得我们在讲生成器的时候,有提及过生成器的状态。同样,在协程这里,我们也了解一下协程(准确的说,应该是Future对象,或者Task任务)有哪些状态。

    Pending:创建future,还未执行
    Running:事件循环正在调用执行任务
    Done:任务执行完毕
    Cancelled:Task被取消后的状态

    可手工 python3 xx.py 执行这段代码,

    import asyncio
    import threading
    import time
    
    async def hello():
        print("Running in the loop...")
        flag = 0
        while flag < 1000:
            with open("F:\test.txt", "a") as f:
                f.write("------")
            flag += 1
        print("Stop the loop")
    
    if __name__ == '__main__':
        coroutine = hello()
        loop = asyncio.get_event_loop()
        task = loop.create_task(coroutine)
    
        # Pending:未执行状态
        print(task)
        try:
            t1 = threading.Thread(target=loop.run_until_complete, args=(task,))
            # t1.daemon = True
            t1.start()
    
            # Running:运行中状态
            time.sleep(1)
            print(task)
            t1.join()
        except KeyboardInterrupt as e:
            # 取消任务
            task.cancel()
            # Cacelled:取消任务
            print(task)
        finally:
            print(task)

    顺利执行的话,将会打印 Pending -> Pending:Runing -> Finished 的状态变化

    假如,执行后 立马按下 Ctrl+C,则会触发task取消,就会打印 Pending -> Cancelling -> Cancelling 的状态变化。

    gather与wait

    还记得上面我说,把多个协程注册进一个事件循环中有两种方法吗?

    • 使用asyncio.wait()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))
    • 使用asyncio.gather()
    # 千万注意,这里的 「*」 不能省略
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.gather(*tasks))

    asyncio.gather 和 asyncio.wait 在asyncio中用得的比较广泛,这里有必要好好研究下这两货。

    还是照例用例子来说明,先定义一个协程函数

    import asyncio
    
    async def factorial(name, number):
        f = 1
        for i in range(2, number+1):
            print("Task %s: Compute factorial(%s)..." % (name, i))
            await asyncio.sleep(1)
            f *= i
        print("Task %s: factorial(%s) = %s" % (name, number, f))

    接收参数方式

    asyncio.wait

    接收的tasks,必须是一个list对象,这个list对象里,存放多个的task。

    它可以这样,用asyncio.ensure_future转为task对象

    tasks=[
           asyncio.ensure_future(factorial("A", 2)),
           asyncio.ensure_future(factorial("B", 3)),
           asyncio.ensure_future(factorial("C", 4))
    ]
    
    loop = asyncio.get_event_loop()
    
    loop.run_until_complete(asyncio.wait(tasks))

    也可以这样,不转为task对象。

    loop = asyncio.get_event_loop()
    
    tasks=[
           factorial("A", 2),
           factorial("B", 3),
           factorial("C", 4)
    ]
    
    loop.run_until_complete(asyncio.wait(tasks))

    asyncio.gather

    接收的就比较广泛了,他可以接收list对象,但是 * 不能省略

    tasks=[
           asyncio.ensure_future(factorial("A", 2)),
           asyncio.ensure_future(factorial("B", 3)),
           asyncio.ensure_future(factorial("C", 4))
    ]
    
    loop = asyncio.get_event_loop()
    
    loop.run_until_complete(asyncio.gather(*tasks))

    还可以这样,和上面的 * 作用一致,这是因为asyncio.gather()的第一个参数是 *coros_or_futures,它叫 非命名键值可变长参数列表,可以集合所有没有命名的变量。

    loop = asyncio.get_event_loop()
    
    loop.run_until_complete(asyncio.gather(
        factorial("A", 2),
        factorial("B", 3),
        factorial("C", 4),
    ))

    甚至还可以这样

    loop = asyncio.get_event_loop()
    
    group1 = asyncio.gather(*[factorial("A" ,i) for i in range(1, 3)])
    group2 = asyncio.gather(*[factorial("B", i) for i in range(1, 5)])
    group3 = asyncio.gather(*[factorial("B", i) for i in range(1, 7)])
    
    loop.run_until_complete(asyncio.gather(group1, group2, group3))

    返回结果不同

    asyncio.wait

    asyncio.wait 返回donespendings

    • dones:表示已经完成的任务
    • pendings:表示未完成的任务

    如果我们需要获取,运行结果,需要手工去收集获取。

    dones, pendings = await asyncio.wait(tasks)
    
    for task in dones:
        print('Task ret: ', task.result())

    wait有控制功能

    import asyncio
    import random
    
    
    async def coro(tag):
        await asyncio.sleep(random.uniform(0.5, 5))
    
    loop = asyncio.get_event_loop()
    
    tasks = [coro(i) for i in range(1, 11)]
    
    
    # 【控制运行任务数】:运行第一个任务就返回
    # FIRST_COMPLETED :第一个任务完全返回
    # FIRST_EXCEPTION:产生第一个异常返回
    # ALL_COMPLETED:所有任务完成返回 (默认选项)
    dones, pendings = loop.run_until_complete(
        asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED))
    print("第一次完成的任务数:", len(dones))
    
    
    # 【控制时间】:运行一秒后,就返回
    dones2, pendings2 = loop.run_until_complete(
        asyncio.wait(pendings, timeout=1))
    print("第二次完成的任务数:", len(dones2))
    
    
    # 【默认】:所有任务完成后返回
    dones3, pendings3 = loop.run_until_complete(asyncio.wait(pendings2))
    
    print("第三次完成的任务数:", len(dones3))
    
    loop.close()
    输出结果
    
    第一次完成的任务数: 1
    第二次完成的任务数: 4
    第三次完成的任务数: 5

    出处:https://www.cnblogs.com/wongbingming/p/9114171.html

  • 相关阅读:
    MongoDB数据库遭大规模勒索攻击,被劫持26000多台服务器 #精选GITHUBMYSQL
    前端追着设计砍系列的9个超酷网页特效
    15款不容错过的前端开发Javascript和css类库
    恶性循环中的永生bug,可以说是相当写实了
    你是码农还是专家?看看你是哪一类程序员
    夏天过去了, 姥爷推荐几套来自smashingmagzine的超棒秋天主题壁纸
    五毛党可能要失业了,因为AI水军来了
    现代软件工程 第五章 【团队和流程】练习与讨论
    现代软件工程 第四章 【结对编程】练习与讨论
    现代软件工程 课件 软件工程师能力自我评价表
  • 原文地址:https://www.cnblogs.com/miaoweiye/p/12021679.html
Copyright © 2011-2022 走看看