async/await套路编程

Posted 2021-01-30 aguncn

对于并发任务，通常是用生成消费模型，对队列的处理可以使用类似master-worker的方式，master主要用户获取队列的msg，worker用户处理消息。

为了简单起见，并且协程更适合单线程的方式，我们的主线程用来监听队列，子线程用于处理队列。这里使用redis的队列。主线程中有一个是无限循环，用户消费队列。

也即：

在主线程里，一个无限循环，一个不断加入的新任务协程：

一个loop.run_forever()，一个async def do_sleep2(x, queue, msg=""):

子线程作消费者。(代码里没有演示，只是用子线程有循环事件和异步加入协程，主线程循环结果)

import time
import asyncio
from queue import Queue
from threading import Thread
from asyncio.futures import Future
from collections.abc import Coroutine, Generator

"""
只要在一个生成器函数头部用上 @asyncio.coroutine 装饰器
就能将这个函数对象，【标记】为协程对象。注意这里是【标记】，划重点。
实际上，它的本质还是一个生成器。
标记后，它实际上已经可以当成协程使用。
"""


@asyncio.coroutine
def hello2():
    yield from asyncio.sleep(1)


coroutine2 = hello2()
print(isinstance(coroutine2, Generator))
print(isinstance(coroutine2, Coroutine))
# True
# False
"""
只要在一个函数前面加上 async 关键字，这个函数对象是一个协程，
通过isinstance函数，它确实是Coroutine类型。
"""


async def hello(name):
    print("Hello, ", name)
    time.sleep(2)
    return ‘stop 2 seconds.‘


# 定义协程对象
coroutine = hello("world")
print(isinstance(coroutine, Coroutine))
# True

# 定义事件循环对象容器
loop = asyncio.get_event_loop()

# 将协程转为task任务
# task = asyncio.ensure_future(coroutine)
task = loop.create_task(coroutine)
print(isinstance(task, Future))
# True
# 将task任务扔进事件循环对象中并触发
loop.run_until_complete(task)

# task.result() 可以取得返回结果
print(‘return value: {}‘.format(task.result()))
# Hello,  world
# return value: stop 2 seconds.


# 协程函数
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)
coroutine4 = do_some_work(4)

# 将协程转成task，并组成list
tasks = [
    asyncio.ensure_future(coroutine1),
    asyncio.ensure_future(coroutine2),
    asyncio.ensure_future(coroutine4),
]

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
# loop.run_until_complete(asyncio.gather(*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 def main():
    coroutine1 = do_some_work(1)
    coroutine2 = do_some_work(2)
    coroutine4 = do_some_work(4)

    tasks = [
        asyncio.ensure_future(coroutine1),
        asyncio.ensure_future(coroutine2),
        asyncio.ensure_future(coroutine4),
    ]

    # 【重点】：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())
# waiting:  1
# waiting:  2
# waiting:  4
# Task ret:  Done after 1s
# Task ret:  Done after 2s
# Task ret:  Done after 4s

"""
协程中的状态

Pending：悬而未决的状态
Running：事件循环正在调用执行任务
Done：任务执行完毕
Cancelled：Task被取消后的状态

asyncio.wait：接收形式少点，控制性强，手工循环结果。

asyncio.gather：接收形式广泛，直接返回结果。
"""


def start_loop(loop):
    # 一个在后台永远运行的事件循环
    asyncio.set_event_loop(loop)
    loop.run_forever()


def start_loop2(loop):
    # 一个在后台永远运行的事件循环
    asyncio.set_event_loop(loop)
    loop.run_forever()


def do_sleep(x, queue, msg=""):
    time.sleep(x)
    queue.put(msg)


async def do_sleep2(x, queue, msg=""):
    await asyncio.sleep(x)
    queue.put(msg)


queue = Queue()
queue2 = Queue()
new_loop = asyncio.new_event_loop()
new_loop2 = asyncio.new_event_loop()

t = Thread(target=start_loop, args=(new_loop, ))
t.start()

t2 = Thread(target=start_loop2, args=(new_loop2, ))
t2.start()

print(time.ctime())

# 动态添加两个协程
# 这种方法，在主线程是同步的
new_loop.call_soon_threadsafe(do_sleep, 6, queue, "第一个")
new_loop.call_soon_threadsafe(do_sleep, 6, queue, "第二个")

# 动态添加两个协程
# 这种方法，在主线程是异步的
asyncio.run_coroutine_threadsafe(do_sleep2(6, queue, "第1个"), new_loop2)
asyncio.run_coroutine_threadsafe(do_sleep2(6, queue, "第2个"), new_loop2)

while True:
    msg = queue.get()
    print("{} 协程运行完成。。。".format(msg))
    print(time.ctime())

# Thu Dec 27 19:51:00 2018
# 第一个 协程运行完成。。。
# Thu Dec 27 19:51:06 2018
# 第二个 协程运行完成。。。
# Thu Dec 27 19:51:12 2018

while True:
    msg = queue2.get()
    print("{} 协程运行完成。。。".format(msg))
    print(time.ctime())

# 第1个 协程运行完成。。。
# Thu Dec 27 20:02:10 2018
# 第2个 协程运行完成。。。
# Thu Dec 27 20:02:10 2018