码迷,mamicode.com
首页 > 其他好文 > 详细

async/await套路编程

时间:2018-12-27 18:22:08      阅读:130      评论:0      收藏:0      [点我收藏+]

标签:后台   call   处理   bsp   dsa   async   并发   col   until   

对于并发任务,通常是用生成消费模型,对队列的处理可以使用类似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

 

async/await套路编程

标签:后台   call   处理   bsp   dsa   async   并发   col   until   

原文地址:https://www.cnblogs.com/aguncn/p/10185094.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!