标签:后台 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
标签:后台 call 处理 bsp dsa async 并发 col until
原文地址:https://www.cnblogs.com/aguncn/p/10185094.html