ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 数据库 > 数据库开发技术 > 一支 asyncio 穿云箭，多个协程来相见

一支 asyncio 穿云箭，多个协程来相见

原创 数据库开发技术 作者：程序员小钟同学 时间：2021-06-04 00:00:00 0 删除 编辑

上一篇中协程主要说的是基于greenle和gevent简单认识，也只是算是从一个初始的认知的协程是什么概念的起步入门。

但是官方推荐的协程库，一直去了解和使用。这一篇大概主要也是讲关于asyncio这个东东。

1 asyncio 异步编程方式

首先asyncio的python的正统的一个协程库，主要是用来对异步IO的支持和实现。

管道定义的协程 通过 async/await 语法进行声明，是编写 asyncio 应用的推荐方式。

asyncio库它包含了几个概念：

  • 异步IO
  • 事件循环
  • 协程函数
  • 协程对象
  • 任务
  • future 概念

对于上述的几个概念，我们需要有所认知，才能更好的去理解的我们的协程的运作的一个流程。

1.1 什么是异步IO？

首先异步IO-通俗的讲就是我们调用的函数里面如果需要执行一个耗时的IO操作时，它只会发出IO指令，并不会等待IO结果返回，此时它会然后就去执行其他代码。等到何时的时机，即当IO返回结果时，再通知CPU进行处理的一种机制。

1.2 什么是事件循环？

首先我们要明白： 事件循环也是一种机制，它是一种处理多并发的一种方式。通常的去理解的话，其实就是一个while循环，里边处理着两个队列的任务，有一个是等待的就绪了的任务列表的，一个已执行完成的任务队列。

事件循环内部执行的是的我们的相关的任务或协程对象。

❝

Ps:通常我们的是把一个协程的对象丢到时间循环里面。

1.3 什么是协程函数和协程对象？

  • 协程函数：所谓的协程函数，不深入概念去说的，其实理解的话，就是一个普通的函数的前面加上一个 async的关键词后，他就是一个协程函数。

  • 协程的对象：其实就是我们的协程函数的实例化的一个对象。意思就是 协程函数（）的就是一个对象。

  • 协程对象和的普通的函数实例化的区别：

    • 普通函数的实例化会调用函数内部的逻辑

    • 协程函数逇实例化不会调用函数内部的逻辑（需要丢到事件循环里面才会去执行）

  • 如何判断函数是否是协程函数的方法：

    • asyncio.iscoroutinefunction(func)     如果 func 是一个 协程函数 则返回 True。

  • 如何判断对象是否是协程对象的方法：

    • asyncio.iscoroutine(obj)     如果 obj 是一个 协程对象 则返回 True。

1.4 什么是Task协程任务？

一些示例说明:

  #!/usr/bin/evn python
# coding=utf-8

import asyncio

async def myfun(time):
    # 模拟耗时的等待任务
    print('我是谁！' + str(time))
    await asyncio.sleep(time)
    return '我是你大爷！' + str(time)


# 第1种写法
async def mytasks_001():
    print('任务列表的生成')
    t1 = asyncio.create_task(myfun(1))
    t2 = asyncio.create_task(myfun(2))
    t3 = asyncio.create_task(myfun(2))
    coroutine_obj1_taskresult = await t1
    coroutine_obj2_taskresult = await t2
    coroutine_obj3_taskresult = await t3
    print('等待任务的列表的执行coroutine_obj1_taskresult',coroutine_obj1_taskresult)


# 第2种写法
async def mytasks1():
    print('任务列表的生成')
    coroutine_obj1_taskresult = await asyncio.create_task(myfun(1))
    coroutine_obj2_taskresult = await asyncio.create_task(myfun(2))
    coroutine_obj3_taskresult = await asyncio.create_task(myfun(3))
    print('等待任务的列表的执行coroutine_obj1_taskresult',coroutine_obj1_taskresult)



# 第2种写法
async def mytasks2():
    print('任务列表的生成')
    # 列表的添加到事件循环中，create_task会立即加入的事件循环里面
    tasks_list = [
        asyncio.create_task(myfun(1)),
        asyncio.create_task(myfun(2)),
        asyncio.create_task(myfun(3))
    ]
    #  通过访问done里的task对象，获取返回值,timeout是整个任务列表执行完成的最长的等待事件，timeout=Node 不限制超时
    done, pending = await asyncio.wait(tasks_list, timeout=None)

    print('列表的执行结果', done)

print('循环事件提交处理-----》')
result = asyncio.run(mytasks_001())
print('循环事件提交处理-----》')
result = asyncio.run(mytasks1())
print('循环事件提交处理-----》')
result = asyncio.run(mytasks2())
print(result)

输出结果（最好自己泡一下）：

循环事件提交处理-----》
任务列表的生成
我是谁！1
我是谁！2
我是谁！2
等待任务的列表的执行coroutine_obj1_taskresult 我是你大爷！1
循环事件提交处理-----》
任务列表的生成
我是谁！1
我是谁！2
我是谁！3
等待任务的列表的执行coroutine_obj1_taskresult 我是你大爷！1
循环事件提交处理-----》
任务列表的生成
我是谁！1
我是谁！2
我是谁！3
列表的执行结果 {<Task finished coro=<myfun() done, defined at C:/Users/15944/Desktop/pyproj/Python/fastapi_aync/协程库/x1.py:38> result='我是你大爷！3'>, <Task finished coro=<myfun() done, defined at C:/Users/15944/Desktop/pyproj/Python/fastapi_aync/协程库/x1.py:38> result='我是你大爷！1'>, <Task finished coro=<myfun() done, defined at C:/Users/15944/Desktop/pyproj/Python/fastapi_aync/协程库/x1.py:38> result='我是你大爷！2'>}
None

从上面可以看得出，Task任务其实是对协程的进一步封装.

  • 在上面的示例的asyncio.wait，如果传入的是协程对象，或协程对象的列表，内会自动的帮你转化为Task对象， 而我们的task又是Future的一个子类，所以我们的asyncio.wait其实处理的就是Future

  • asyncio.wait返回的是一个元祖对象，我们可以通过访问done里的task对象，获取返回值

  • timeout是整个任务列表执行完成的最长的等待事件，timeout=Node 不限制超时

❝

关于任务对象的创建，不同的版本有不同的方式：

  • 3.7之前的版本 使用：asyncio.ensure_future（）
  • 3.7后可以使用：asyncio.creat_task(协程对象)
❝

关于任务的添加：

loop.create_task 只是创建任务, 不会立即执行

loop = asyncio.get_event_loop()      # 创建一个事件loop loop.run_until_complete(main(loop))  # --真的执行任务的地方是将事件加入到事件循环loop

一个使用asyncio.ensure_future方式提交的示例：

import asyncio


async def myfun(time):
    # 模拟耗时的等待任务
    print('我是谁！' + str(time))
    await asyncio.sleep(time)
    return '我是你大爷！' + str(time)


loop = asyncio.get_event_loop()
task = asyncio.ensure_future(myfun(2))
result = loop.run_until_complete(task)
print('返回的结果:', result)

1.4.1 Task的状态

Task的状态：pending 和 finished

loop = asyncio.get_event_loop()
task = loop.create_task(main())
print(task)  # pending状态的Task
loop.run_until_complete(task)
print(task)     # finished状态的Task

因为我们的Task是Future的子类，而我们的Future的状态有多种：

  • Pending

  • Running

  • Done

  • Cancelled

  • FINISHED

这个地方可以通过查看Future源码可以知道！

大概查看示例：

import asyncio


async def myfun(time):
    # 模拟耗时的等待任务
    print('我是谁！' + str(time))
    await asyncio.sleep(time)
    return '我是你大爷！' + str(time)


loop = asyncio.get_event_loop()
task = asyncio.ensure_future(myfun(2))
print(task.done())
print('协程任务的状态:',task._state)

print("是否取消：",task.cancelled())

result = loop.run_until_complete(task)
print("是否已经：",task.done())
print(task.done())
print('协程任务的状态:',task._state)
# 在 _state 为 FINISHED 时，可以直接result 值
print("协程任务的结果---返回的还是一个协程对象：",task.result)
print("协程任务的结果---需要（）：",task.result())
print('返回的结果:', result)

输出结果：

False
协程任务的状态: PENDING
是否取消： False
我是谁！2
是否已经： True
True
协程任务的状态: FINISHED
协程任务的结果---返回的还是一个协程对象： <built-in method result of _asyncio.Task object at 0x000001F4AACF6048>
协程任务的结果---需要（）： 我是你大爷！2
返回的结果: 我是你大爷！2

1.4.1 Task的结果的绑定回调 

import asyncio


async def myfun(time):
    # 模拟耗时的等待任务
    print('我是谁！' + str(time))
    await asyncio.sleep(time)
    return '我是你大爷！' + str(time)

def mytasjk_call_back(future):
    print('回调的函数的结果的',id(future))
    print("回调回调函数的的结果: {}".format(future.result()))

loop = asyncio.get_event_loop()
task = asyncio.ensure_future(myfun(2))
# 添加结果的回调
task.add_done_callback(mytasjk_call_back)
print('任务DUIX id', id(task))
result = loop.run_until_complete(task)
print('结构的ID id', id(result))
print('返回的结果:', result)
# 不同的方式获取返回的额结果！
print('返回的结果:', task.result())
print('返回的结果:', result)


1.5 什么是future？

asyncio.Future类：代表将来执行或没有执行的任务计算的结果,它与task没有本质的区别。按官网说法说它是一个特殊的低层级的可等待对象，通常情况，没必要创建Future对象。

1.6 async/await关键字一些理解

async/await 关键字：python3.5用于定义协程的关键字，

  • async 定义一个协程函数

  • await用于挂起阻塞的异步调用接口,可以理解为等待IO结果的返回。

❝

  • await 遇到IO后，此时协程对（协程任务）象就会被事件循环给挂起，循环转而执行其他协程对象（协程任务），非IO的话不一定会挂起

  • 如果await 后面是协程对象，则会执行直接的执行这个协程

  • await + 可等待对象 . 可等待对象有三种主要类型: 协程, 任务 和 Future.

  • 如果一个对象可以在 await 语句中使用，那么它就是 可等待 对象。许多 asyncio API 都被设计为接受可等待对象。

wait 后面三个可能对象的示例：

import asyncio

async def test_time(time):
    print("耗时任务{}开始导入！{}".format(time,time))
    await asyncio.sleep(time)
    return '我耗时多少才处理完{}'.format(time)



async def main():
    # 下面的三种写法都可以
    # 协程对象的方式
    tasks =[test_time(i) for i in range(1,4)]
    # ensure_future方式
    # tasks = [ asyncio.ensure_future(test_time(i)) for i in range(1, 4)]
    # create_task 方式
    # tasks = [ asyncio.create_task(test_time(i)) for i in range(1, 4)]
    for fut in asyncio.as_completed(tasks):
        earliest_result = await fut
        # 会依次打印 bar 和 foo, 因为 bar() 会更早执行完毕
        print(earliest_result)


if __name__ == '__main__':
    import time
    now = time.time()
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(main())
    finally:
        loop.close()
    print("总用时", time.time() - now)

1.7 单一简单的协程示例

运行一个协程的基本流程是：

  • 1：通过async关键字定义一个需要执行的协程函数
  • 2：实例化我们的协程函数为一个实例对象
  • 3：创建一个事件循环实例，也可以说是协程对象的调度器（3.7以前的版本）
  • 4：把协程对象放到我们的事件循环中并启动事件循环来执行
  • 5：可以获取的处理的协程任务的结果的回调

import asyncio


# 第一步：通过async关键字定义一个需要执行的协程函数
async def myfun(time):
    # 模拟耗时的等待任务
    print('我是谁！' + str(time))
    await asyncio.sleep(time)
    return '我是你大爷！'


# 第二步：实例化我们的协程函数为一个实例对象
coroutine_obj = myfun(1)  # 协程是一个对象，不能直接运行

# 第三步： 创建一个事件循环实例，也可以说是协程对象的调度器（3.7以前的版本）
loop = asyncio.get_event_loop()
#
# 第四步：把协程对象放到我们的事件循环中并执行
result = loop.run_until_complete(coroutine_obj)
# 可以获取的处理的协程任务的结果的回调
print(result)

❝

PS:3.7后我们的 asyncio.run(coroutine_obj) ==

# 第三步： 创建一个事件循环实例，也可以说是协程对象的调度器（3.7以前的版本）
loop = asyncio.get_event_loop()
#
# 第四步：把协程对象放到我们的事件循环中并执行
result = loop.run_until_complete(coroutine_obj)
❝

asyncio.run() 函数用来运行最高层级的入口点,运行我们的协程对象

1.8 链路协程示例

链路示例其实就是等待之中还有等待的链路连接起来。就是一个协程函数等待另一个协程函数。

import asyncio


# 第一步：通过async关键字定义一个需要执行的协程函数
async def myfun(time):
    # 模拟耗时的等待任务
    print('你是谁！' + str(time))
    return '你是谁！' + str(time)


async def myfun2(time):
    # 模拟耗时的等待任务
    print('我是你大爷的大爷！' + str(time))
    return '我是你大爷的大爷！' + str(time)


async def pacthfun(time):
    # 模拟耗时的等待任务
    result1 = await myfun(1)
    await asyncio.sleep(time)
    result2 = await myfun2(result1)
    return result2


# 第二步：实例化我们的协程函数为一个实例对象
coroutine_obj = pacthfun(1)  # 协程是一个对象，不能直接运行

# 第三步： 创建一个事件循环实例，也可以说是协程对象的调度器（3.7以前的版本）

event_loop = asyncio.get_event_loop()
try:
    # 第四步：把协程对象放到我们的事件循环中并执行
    return_value = event_loop.run_until_complete(coroutine_obj)
    # 可以获取的处理的协程任务的结果的回调
    print('result value: {!r}'.format(return_value))
finally:
    event_loop.close()

1.9 多协程任务示例-run_until_complete方式 

import asyncio


# 第一步：通过async关键字定义一个需要执行的协程函数
async def myfun(time):
    # 模拟耗时的等待任务
    print('我是谁！' + str(time))
    await asyncio.sleep(time)
    return '我是你大爷！' + str(time)


# 第二步：实例化我们的协程函数为一个实例对象
coroutine_obj1 = myfun(1)  # 协程是一个对象，不能直接运行
coroutine_obj2 = myfun(2)  # 协程是一个对象，不能直接运行
coroutine_obj3 = myfun(3)  # 协程是一个对象，不能直接运行
coroutine_obj4 = myfun(4)  # 协程是一个对象，不能直接运行
# 第三步： 创建一个事件循环实例，也可以说是协程对象的调度器（3.7以前的版本）
loop = asyncio.get_event_loop()
#
# 第四步：把协程对象放到我们的事件循环中并执行
result = loop.run_until_complete(asyncio.gather(coroutine_obj1, coroutine_obj2, coroutine_obj3, coroutine_obj4))

# 获取协程对象并没有返回结果-asyncio.gather执行的结果返回的是一个列表
print(result)
# 保持我们的事件循环一直在运行中，
loop.run_forever()

输出结果：

我是谁！1
我是谁！2
我是谁！3
我是谁！4
['我是你大爷！1', '我是你大爷！2', '我是你大爷！3', '我是你大爷！4']

1.10 协程并发执行

并发执行，其实是一场常见需求，如组合批量请求，且需要等待返回的时候。这种其实就是:

一个的协程任务--需要等待 批量多个的协程任务的完成。

1.10.1 asyncio.gather··并发的执行协程

  • 按顺序的提交，一旦调用就是，执行并等待所有任务完成！

  • asyncio.gather（* aws，loop = None，return_exceptions = False ）

 注意点：

 * gather会等待最耗时的那个完成之后才返回结果，耗时总时间取决于其中任务最长时的协程

 * aws 带传入的协程的对象，或协程队列的列表，或者是一个future,如果传入的是协程类型的，它将自动调度为任务Task.

 >PS 两者的区别提交： 有* 和没 *
❝
  • await asyncio.gather(*[myWorker(lock, 1), myWorker(lock, 2), myWorker(lock, 3)])
  • await asyncio.wait([myWorker(lock, 1), myWorker(lock, 2), myWorker(lock, 3)])

1.10.2 asyncio.wait()··并发的执行协程

  • 乱序的执行，一旦调用就是，执行并等待所有任务完成！
❝

PS 两者的区别提交： 有* 和没 *

  • await asyncio.gather(*[myWorker(lock, 1), myWorker(lock, 2), myWorker(lock, 3)])
  • await asyncio.wait([myWorker(lock, 1), myWorker(lock, 2), myWorker(lock, 3)])

    • wait 返回一个由两个集合组成的元祖，
  • 一个保存状态为 done 的 Task ，一个保存状态为 pending 的 Task:
  • 一个集合来保存它创建的 Task 实例
  • 因为执行的无序，所以保存的 Task 的结果也是无序的

1.10.3 asyncio.as_completed()··并发的执行协程

  • asyncio.as_completed(aws, ***, loop=None, timeout=None)

    并发地运行 aws 集合中的 可等待对象。 返回一个协程的迭代器。 所返回的每个协程可被等待以从剩余的可等待对象集合中【获得最早】的下一个结果，意思是哪个先处理完成的我先返回哪个！！！

❝

PS:意思就是返回的结果集，我是先把优先处理完成的先取出来！

import asyncio

async def test_time(time):
    print("耗时任务{}开始导入！{}".format(time,time))
    await asyncio.sleep(time)
    return '我耗时多少才处理完{}'.format(time)



async def main():
    # 下面的三种写法都可以
    # 协程对象的方式
    tasks =[test_time(i) for i in range(1,4)]
    # ensure_future方式
    # tasks = [ asyncio.ensure_future(test_time(i)) for i in range(1, 4)]
    # create_task 方式
    # tasks = [ asyncio.create_task(test_time(i)) for i in range(1, 4)]
    for fut in asyncio.as_completed(tasks):
        earliest_result = await fut
        # 会依次打印 bar 和 foo, 因为 bar() 会更早执行完毕
        print(earliest_result)


if __name__ == '__main__':
    import time
    now = time.time()
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(main())
    finally:
        loop.close()
    print("总用时", time.time() - now)

输出结果：

耗时任务1开始导入！1
耗时任务3开始导入！3
耗时任务2开始导入！2
我耗时多少才处理完1
我耗时多少才处理完2
我耗时多少才处理完3
总用时 2.997316598892212

1.11 loop.run_forever()

run_forever 顾名思义，就是一直运行下去，起一个死循环的方式，一直简单的我们的事件循环里面的任务。

1.12 协程任务的取消

来自官网的示例：

import asyncio


async def cancel_me():
    print('cancel_me(): before sleep')

    try:
        # Wait for 1 hour
        await asyncio.sleep(3600)
    except asyncio.CancelledError:
        print('cancel_me(): cancel sleep')
        raise
    finally:
        print('cancel_me(): after sleep')


async def main():
    # Create a "cancel_me" Task
    task = asyncio.create_task(cancel_me())

    # Wait for 1 second
    await asyncio.sleep(1)

    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("main(): cancel_me is cancelled now")


asyncio.run(main())

2 协程同步问题

但凡有并发通常都会涉及到的同步问题。虽然我们的协程是在用单线程下运行，但是它还是并发的，乱序执行的，主要的原因是：我们的协程执行的时候会存在延时或者I/O中断等因素，所以为了实现每个协程的同步，还是得使用一些同步原语来实现。 协程也有一套类似的线程的同步的机制。

官网的说法是：

2.1 协程锁-lock

asyncio 在的锁其实也是一个互斥锁，一个锁对象Lock，它一次只允许一个协程来访问共享的资源，如果多协程想访问就会阻塞起来，也就是说如果一个协程没有释放这个锁，别的协程是没有办法访问共享的资源。

import asyncio


async def qiangpaio(lock, name):
    print(f"{name}想要获得抢票！")
    async with  lock:
        print(f"上锁了！{name}抢到了票！！给它处理出票手续！")
        await asyncio.sleep(3)
    print(f"{name}出票手续办完了！让出锁！给别人！")



async def main():
    # 定义一写成所
    lock = asyncio.Lock()
    # 并发的提交执行所有的协程---这里使用的gather的方式
    await asyncio.gather(*[qiangpaio(lock, '小明:'), qiangpaio(lock, '小王：'), qiangpaio(lock, '小看：')])
    # wait 是乱序的一种方式
    # await asyncio.wait([myWorker(lock, 1), myWorker(lock, 2), myWorker(lock, 3)])


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    try:
        # 执行协程任务
        loop.run_until_complete(main())
        print("所有抢票人都抢完成")
    except KeyboardInterrupt as e:
        print('当前协程里面的所有任务：',asyncio.Task.all_tasks())
        print('批量取消里面所有的额任务：', asyncio.gather(*asyncio.Task.all_tasks()).cancel())
        # 停止事件循环
        # loop.stop()
        # loop.run_forever()
    finally:
        # 关闭事件循环
        loop.close()

输出结果：

小明:想要获得抢票！
上锁了！小明:抢到了票！！给它处理出票手续！
小王：想要获得抢票！
小看：想要获得抢票！
小明:出票手续办完了！让出锁！给别人！
上锁了！小王：抢到了票！！给它处理出票手续！
小王：出票手续办完了！让出锁！给别人！
上锁了！小看：抢到了票！！给它处理出票手续！
小看：出票手续办完了！让出锁！给别人！
所有抢票人都抢完成
❝

上锁和解锁的方式的换成： with await lock:

2.2 协程队列（Queue和LifoQueue和PriorityQueue）

2.2.1 Queue队列示例

队列一些基础说明：

常见的一些异常：
    - exception asyncio.QueueEmpty
    当队列为空的时候，调用 get_nowait() 方法而引发这个异常。
    - exception asyncio.QueueFull
    当队列中条目数量已经达到它的 maxsize 的时候，调用 put_nowait() 方法而引发的异常。

2.2.1.1对比线程队列处理：

线程队列

from queue import Queue
import time, threading


class Producer(threading.Thread):

    def run(self):
        ssds = "你是一个大爷"
        import random
        while True:
            item = random.choice(ssds)
            queue.put(item)
            print('写入元素: %s' % item)
            print('当前队列内容: %s' % queue.queue)
            time.sleep(1)


class Consumer(threading.Thread):

    def run(self):
        while True:
            try:
                item = queue.get(block=False, timeout=1)
                print("=====>" * 10)
                print('从队列中获取元素: %s' % item)
                print("=====>" * 10)
                time.sleep(1)
                queue.task_done()
            except:
                pass



if __name__ == '__main__':
    # queue = ThreadSafeQueue(max_size=100)
    queue = Queue(maxsize=100)

    producer = Producer()
    producer2 = Producer()
    consumer = Consumer()

    producer.start()
    producer2.start()

    consumer.start()
    producer.join()
    producer2.join()
    consumer.join()

协程队列的示例：


import asyncio
import random

class Pro():

    async def runaction(self,queue):
        ssds = "你是一个大爷"

        while True:

            item = random.choice(ssds)
            await queue.put(item)
            print('写入元素: %s' % item)
            print('当前队列内容: %s' % queue)
            await asyncio.sleep(random.random())

class Consumer():

    async def runaction(self,queue):
        print("啥还是兑换")
        while True:
            try:
                item = await queue.get()
                print("=====>" * 10)
                print('从队列中获取元素: %s' % item)
                print("=====>" * 10)
                await asyncio.sleep(1)
                queue.task_done()
            except:
                print("ds")
                pass

async def run(n):
    # 实例化一个队列对象---注意区别于线程的
    queue = asyncio.Queue()
    producer = Pro()
    consumer= Consumer()

    consumer_result = asyncio.ensure_future(consumer.runaction(queue))
    await producer.runaction(queue)
    # 等到消费者消费完所有任务
    await queue.join()
    # 消费任务取消
    consumer_result.cancel()

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run(10))
    loop.close()

协程结果输出：

写入元素: 你
当前队列内容: <Queue maxsize=0 _queue=['你'] tasks=1>
啥还是兑换
=====>=====>=====>=====>=====>=====>=====>=====>=====>=====>
从队列中获取元素: 你
=====>=====>=====>=====>=====>=====>=====>=====>=====>=====>
写入元素: 爷
当前队列内容: <Queue maxsize=0 _queue=['爷'] tasks=2>
=====>=====>=====>=====>=====>=====>=====>=====>=====>=====>
从队列中获取元素: 爷
=====>=====>=====>=====>=====>=====>=====>=====>=====>=====>
写入元素: 个
当前队列内容: <Queue maxsize=0 _queue=['个'] tasks=2>
写入元素: 你
当前队列内容: <Queue maxsize=0 _queue=['个', '你'] tasks=3>
写入元素: 大
当前队列内容: <Queue maxsize=0 _queue=['个', '你', '大'] tasks=4>
=====>=====>=====>=====>=====>=====>=====>=====>=====>=====>
从队列中获取元素: 个

2.2.1.2协程的Queue结合aiohttp的简单异步请求示例 

import asyncio
import random
import aiohttp
import async_timeout



class Pro():

    async def runaction(self,queue):
        index = 0
        while True:
            index = index+1
            item = "http://www.baidu.com?q={}".format(index)
            await queue.put(item)
            print('写入元素: %s' % item)
            print('当前队列内容: %s' % queue)
            await asyncio.sleep(random.random())

class Consumer():


    headers = {'user-agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/65.0.3325.181 Safari/537.36'}

    async def runaction(self,queue):
        # while not queue.empty():
        while True:
            try:
                item = await queue.get()
                print("=====>" * 10)
                print('从队列中获取元素: %s' % item)
                print("=====>" * 10)
                rsu=await self.get_body(item)
                print('结果：',rsu)
                queue.task_done()
            except:
                print("ds")
                pass

    async def get_body(self,url):
        # 一个异步的请求的回话对象
        async with aiohttp.ClientSession() as session:
            try:
                # 设置请求的操作
                with async_timeout.timeout(10):
                    # 提交的请求
                    async with session.get(url, headers=self.headers) as response:
                        # 判断请求的状态---判断状态是否正常
                        response.raise_for_status()
                        # 判断状态的值
                        if response.status == 200:
                            # 异步的解析的返回我们的内容信息-await后面接的需要是一个可调用的：协程对象，或任务对象或对象
                            html = await response.text()
                            return html
                        else:
                            return {'errorcode': response.status, 'html': ''}
            except Exception as err:
                return {'errorcode': err, 'html': ''}

async def run():
    # 实例化一个队列对象---注意区别于线程的
    queue = asyncio.PriorityQueue()
    producer = Pro()
    consumer= Consumer()

    consumer_result = asyncio.ensure_future(consumer.runaction(queue))
    await producer.runaction(queue)
    # 等到消费者消费完所有任务
    await queue.join()
    # 消费任务取消
    consumer_result.cancel()

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(run())
    loop.close()

其他优先级队列和现金后出的队列，这里不复述！！！可以参考线程的示例@

2.3 协程条件变量-Event

官网的示例：

async def waiter(event):
    print('waiting for it ...')
    await event.wait()
    print('... got it!')

async def main():
    # Create an Event object.
    event = asyncio.Event()

    # Spawn a Task to wait until 'event' is set.
    waiter_task = asyncio.create_task(waiter(event))

    # Sleep for 1 second and set the event.
    await asyncio.sleep(1)
    event.set()

    # Wait until the waiter task is finished.
    await waiter_task

asyncio.run(main())

之前类似线程的示例：

import asyncio
import random
event = asyncio.Event()


async def restaurant_handle(event):

    print("肯德基开门了！！")
    # print("【服务员-1】当前事件状态：",event.is_set())
    print("1、【服务员】请问你要吃点什么吗？。。。")
    await asyncio.sleep(random.randint(1,3))
    event.set()#解除阻塞状态
    #清除已有的状态
    # 当调用wait（）方法是将进入到阻塞状态，同时会设置阻塞标记为“False”，(待阻塞标记为“True"后才会接触阻塞状态）
    await event.wait()#等待客人点餐之后，才能进行下一步的后续处理

    print("3、【服务员】你自己一个人这么能吃啊！。。。")
    await asyncio.sleep(random.randint(1,3))
    event.set() #解除阻塞状态
      # 之前的状态清空
    await event.wait()

    print("5、【服务员】好的，我现在就给你下单，请你稍等。。。")
    await asyncio.sleep(random.randint(1,3))
    event.set()  # 解除阻塞状态
      # 之前的状态清空
    await event.wait()

    print("7、【服务员】好的，你稍等一下~~~。。。")
    await asyncio.sleep(random.randint(1,3))
    event.set()
    # 
    # await event.wait()

    print("8、【服务员】客人你点的：一个鸡腿+一个汉堡+一个这么大的鸡排 已做好了！请慢慢享用！")
    await asyncio.sleep(random.randint(1,3))
    # event.set()
    # 
    await event.wait()

    print("11、【服务员】好的，总共消费1000快！。。。")
    await asyncio.sleep(random.randint(1, 3))
    event.set()  # 解除阻塞状态
      # 之前的状态清空
    await event.wait()

async def diners_hangle(event):#食客的处理进程

    event.set()  # 解除阻塞状态
    print("当前事件状态：", event.is_set())
    await event.wait() #等待之前的第一步完成  两个进程所以先阻塞，让另一个执行

    print("2、【客人】我要点一个鸡腿+一个汉堡+一个这么大的鸡排！。。。")
    await asyncio.sleep(random.randint(1,3))
    event.set() #解除阻塞状态
    #之前的状态清空
    await event.wait()#继续等待后续的处理步骤

    print("4、【客人】怎么不可以吗？快下单吧~我都饿的叽里咕噜了~")
    await asyncio.sleep(random.randint(1,3))
    event.set()
    
    await event.wait()

    print("6、【客人】让师傅做快一点。。。")
    await asyncio.sleep(random.randint(1,3))
    event.set()
    
    await event.wait()

    print("9、【客人】好的，我开始吃了~~~~")
    # event.set()
    await asyncio.sleep(random.randint(1, 3))

    print("10、【客人】吃完了！服务员，我买单！！！~~~~")
    await asyncio.sleep(random.randint(1, 3))
    event.set()
    
    await event.wait()

    print("12、【客人】这么黑~~1000快！！不给！！我跑了！再见！~~~~")
    event.set()


async def main():
    # Spawn a Task to wait until 'event' is set.
    waiter_task = asyncio.create_task(diners_hangle(event))


    await restaurant_handle(event)
    # 等到消费者消费完所有任务

    # asyncio.create_task(diners_hangle(event))
    # Sleep for 1 second and set the event.
    await asyncio.sleep(1)
    # event.set()
    # Wait until the waiter task is finished.
    await waiter_task


if __name__ == '__main__':
    asyncio.run(main())

输出结果：

肯德基开门了！！
1、【服务员】请问你要吃点什么吗？。。。
当前事件状态： True
2、【客人】我要点一个鸡腿+一个汉堡+一个这么大的鸡排！。。。
4、【客人】怎么不可以吗？快下单吧~我都饿的叽里咕噜了~
6、【客人】让师傅做快一点。。。
3、【服务员】你自己一个人这么能吃啊！。。。
9、【客人】好的，我开始吃了~~~~
5、【服务员】好的，我现在就给你下单，请你稍等。。。
10、【客人】吃完了！服务员，我买单！！！~~~~
12、【客人】这么黑~~1000快！！不给！！我跑了！再见！~~~~

上面的执行的是乱序的！

这种条件变量-Condition其实和Event类似：

2.4 协程条件变量-Condition

首先协程的Condition的条件变量，他不是线程安全的。条件变量主要可被任务用于等待某个事件发生，然后获取对共享资源的独占访问。

这种条件变量-Condition其实和Event类似

下面文字来自官网的说明：

  • 在本质上，Condition 对象合并了 Event 和 Lock 的功能
  • 多个 Condition 对象有可能共享一个 Lock，这允许关注于共享资源的特定状态的不同任务实现对共享资源的协同独占访问。

示例来源： https://mozillazg.com/2017/08/python-asyncio-note-synchronization-primitives.html

简单示例：


import asyncio


async def consumer(condition, n):
    # 7：协程任务的就绪等待
    async with condition:
        print('生产者 {} 正在等待'.format(n))
        await condition.wait()
        print('生产者 {} 开始触发'.format(n))
    print('结束 consumer {}'.format(n))


async def manipulate_condition(condition):
    print('开始生产产品')

    # 5：暂停让消费者开始----遇到阻塞---就会切换到consumer
    await asyncio.sleep(0.1)

    print('任务已添加成功完毕！')
    print('---->' * 10)
    for i in range(1, 3):
        # 随机的通知协程对象数
        async with condition:
            print('随机的通知协程对象数--通知协程对象消费者 {} consumers'.format(i))
            # 唤醒最多 n 个正在等待此条件的任务（默认为 1 个）
            condition.notify(n=i)
        await asyncio.sleep(0.1)

    # 通知剩余的消费者信息
    print('---->' * 10)
    async with condition:
        print('通知剩余消费者协程任务')
        condition.notify_all()

    print('结束！！！！！！！')


async def main(loop):
    # 1：创建一个条件对象。 该对象不是线程安全的。
    condition = asyncio.Condition()
    # 2:设置任务监听的条件
    consumers = [
        consumer(condition, i)
        for i in range(5)
    ]

    # 3：提交操作条件变量的任务协程--创建任务协程的对象，并添加到事件循环中
    loop.create_task(manipulate_condition(condition))

    # 4：等待所有的协程任务的完成
    await asyncio.wait(consumers)

# 获取循环的对象
event_loop = asyncio.get_event_loop()
try:
    # 执行协程异步任务
    result = event_loop.run_until_complete(main(event_loop))
finally:
    event_loop.close()

输出结果：

开始生产产品
生产者 3 正在等待
生产者 1 正在等待
生产者 2 正在等待
生产者 0 正在等待
生产者 4 正在等待
任务已添加成功完毕！
---->---->---->---->---->---->---->---->---->---->
随机的通知协程对象数--通知协程对象消费者 1 consumers
生产者 3 开始触发
结束 consumer 3
随机的通知协程对象数--通知协程对象消费者 2 consumers
生产者 1 开始触发
结束 consumer 1
生产者 2 开始触发
结束 consumer 2
---->---->---->---->---->---->---->---->---->---->
通知剩余消费者协程任务
结束！！！！！！！
生产者 0 开始触发
结束 consumer 0
生产者 4 开始触发
结束 consumer 4
  • 1：创建一个条件对象。 该对象不是线程安全的。
  • 2：设置任务监听的条件，启动了五个 Condition 的消费者，每个都使用 wait() 方法来等待它们可以继续处理的通知。
  • 3：提交操作条件变量的任务协程--创建任务协程的对象，并添加到事件循环中，manipulate_condition() 首先通知了一个消费者，然后有通知了两个消费者，最后通知剩下的所有消费者。
  • 4：等待所有的协程任务的完成
  • 5：协程任务的结束退出

2.5 协程信号量（并发量）-

  • 信号量其实就是并发任务池的最大数的意思，它可以控制协程并发量的池的数。
import time, asyncio, aiohttp

url = 'https://www.baidu.com/'


async def action(url, semaphore):
    # 限制批量的并发执行的任务数
    async with semaphore:
        # 定义session回话
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                # 判断请求的状态---判断状态是否正常
                response.raise_for_status()
                # 判断状态的值
                print("当前返回的状态码", response.status)
                if response.status == 200:
                    # 异步的解析的返回我们的内容信息-await后面接的需要是一个可调用的：协程对象，或任务对象或对象
                    html = await response.text()
                    # text()返回字符串数据；
                    # read()返回二进制数据
                    # json()返回json对象
                    print("返回的结果：", html[2, 50])
                    return html


async def main():
    # 定义可以同时进行的并发的协程任务数500，
    semaphore = asyncio.Semaphore(2)
    # 生成总的可以执行的任务数100000
    to_get = [action(url.format(), semaphore) for _ in range(100000)]
    # 执行批量的并发
    await asyncio.wait(to_get)


loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

资料参考

https://docs.python.org/zh-cn/3/library/asyncio-task.html#coroutine

https://blog.csdn.net/wucong60/article/details/102216056

❝

个人其他博客地址

❝

简书：https://www.jianshu.com/u/d6960089b087

掘金：https://juejin.cn/user/2963939079225608

小钟同学 | 文 【原创】| QQ：308711822

  • 1：本文相关描述主要是个人的认知和见解，如有不当之处，还望各位大佬指正。
  • 2：关于文章内容，有部分内容参考自互联网整理，如有链接会声明标注；如没有及时标注备注的链接的，如有侵权请联系，我会立即删除处理哟。


原文作者：程序员小钟同学；如有侵权，请联系管理员删除。

0
0
上一篇： DotMap 操作原生字典类型转为类对象操作的库
下一篇： python 并发编程-协程篇（2）greenle和gevent简单认识
请登录后发表评论 登录
全部评论
程序员小钟同学

注册时间：2021-02-20

  • 博文量
    224
  • 访问量
    111765

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593