如何在Python中编写自己的异步/等待协程函数?

本文介绍了如何在Python中编写自己的异步/等待协程函数?的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

我正在尝试编写自己的awaiatbale函数，该函数可以在asyncio循环中使用，例如asyncio.sleep()方法或类似这些可等待的实现方法.

I'm trying to write my own awaiatbale function which could use in asyncio loop such as asyncio.sleep() method or something like these pre-awaitable implemented methods.

这是我到目前为止所做的:

Here is what I've done so far:

import asyncio def coro1(): for i in range(1, 10): yield i def coro2(): for i in range(1, 10): yield i*10 class Coro: # Not used. def __await__(self): for i in range(1, 10): yield i * 100 @asyncio.coroutine def wrapper1(): return (yield from coro1()) @asyncio.coroutine def wrapper2(): return (yield from coro2()) for i in wrapper1(): print(i) print("Above result was obvious which I can iterate around a couroutine.".center(80, "#")) async def async_wrapper(): await wrapper1() await wrapper2() loop = asyncio.get_event_loop() futures = [asyncio.ensure_future(async_wrapper())] result = loop.run_until_complete(asyncio.gather(*futures)) print(result) loop.close()

我得到的结果是

1 2 3 4 5 6 7 8 9 #######Above result was obvious which I can iterate around a couroutine.######### Traceback (most recent call last): File "stack-coroutine.py", line 36, in <module> result = loop.run_until_complete(asyncio.gather(*futures)) File "/usr/lib/python3.6/asyncio/base_events.py", line 484, in run_until_complete return future.result() File "stack-coroutine.py", line 30, in async_wrapper await wrapper1() File "stack-coroutine.py", line 18, in wrapper1 return (yield from coro1()) File "stack-coroutine.py", line 5, in coro1 yield i RuntimeError: Task got bad yield: 1

我期望得到的结果:

1 10 2 20 3 30 . . .

[注意]:

• 我不是在寻找多线程或多进程方法.
• 这个问题与我尚未解决的问题几乎相似./li>
• 我正在使用 Python3.6

• I'm not looking for a multithread or multiprocess method.
• This Question is almost similar to my question which has not resolved yet.
• I'm using Python3.6

推荐答案

我发现了使用生成器的并发/异步方法.但是，这不是asyncio方法:

I found a concurrency/asynchronous approach using generators. However, it is not an asyncio approach:

from collections import deque def coro1(): for i in range(1, 5): yield i def coro2(): for i in range(1, 5): yield i*10 print('Async behaviour using default list with O(n)'.center(60, '#')) tasks = list() tasks.extend([coro1(), coro2()]) while tasks: task = tasks.pop(0) try: print(next(task)) tasks.append(task) except StopIteration: pass print('Async behaviour using deque with O(1)'.center(60, '#')) tasks = deque() tasks.extend([coro1(), coro2()]) while tasks: task = tasks.popleft() # select and remove a task (coro1/coro2). try: print(next(task)) tasks.append(task) # add the removed task (coro1/coro2) for permutation. except StopIteration: pass

出局:

########Async behaviour using default list with O(n)######## 1 10 2 20 3 30 4 40 ###########Async behaviour using deque with O(1)############ 1 10 2 20 3 30 4 40

[更新]:

[UPDATE]:

最后，我已经通过AsyncIO语法解决了这个示例:

Finally, I've solved this example through AsyncIO syntax:

import asyncio async def coro1(): for i in range(1, 6): print(i) await asyncio.sleep(0) # switches task every one iteration. async def coro2(): for i in range(1, 6): print(i * 10) await asyncio.sleep(0) # switches task every one iteration. loop = asyncio.get_event_loop() futures = [ asyncio.ensure_future(coro1()), asyncio.ensure_future(coro2()) ] loop.run_until_complete(asyncio.gather(*futures)) loop.close()

出局:

1 10 2 20 3 30 4 40 5 50

以及通过async-await表达式和基于 堆队列算法，而无需使用asyncio库及其事件循环以及asyncio.sleep()方法:

And another concurrency coroutine manner via async-await expression and an event-loop manager based on Heap queue algorithm, without using asyncio library and its event-loop and without asyncio.sleep() method:

import heapq from time import sleep from datetime import datetime, timedelta class Sleep: def __init__(self, seconds): self.sleep_until = datetime.now() + timedelta(seconds=seconds) def __await__(self): yield self.sleep_until async def coro1(): for i in range(1, 6): await Sleep(0) print(i) async def coro2(): for i in range(1, 6): await Sleep(0) print(i * 10) def coro_manager(*coros): coros = [(datetime.now(), coro) for coro in coros] heapq.heapify(coros) while coros: exec_at, coro = heapq.heappop(coros) if exec_at > datetime.now(): sleep((exec_at - datetime.now()).total_seconds()) try: heapq.heappush(coros, (coro.send(None), coro)) except StopIteration: try: coros.remove(coro) except ValueError: pass coro_manager(coro1(), coro2())

出局:

1 10 2 20 3 30 4 40 5 50