blocks|key|1797970|text|async/await的重点是交叉任务，而不是函数/生成器。例如，当您await+asyncio.sleep(1)时，您当前的协同作用会随着睡眠而延迟。类似地，async+for将其协同线延迟到下一项准备就绪。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|1797971|为了运行独立的功能，必须将每个部分创建为一个单独的任务。使用Queue在它们之间交换项--任务只会延迟到它们交换了项为止。|1797972|from+asyncio+import+Queue,+sleep,+run,+gather


#+the+original+async+generator
async+def+g():
++++for+i+in+range(3):
++++++++await+sleep(1)
++++++++yield+i


async+def+producer(queue:+Queue):
++++async+for+i+in+g():
++++++++print('send',+i)
++++++++await+queue.put(i)++#+resume+once+item+is+fetched
++++await+queue.put(None)


async+def+consumer(queue:+Queue):
++++x+=+await+queue.get()++#+resume+once+item+is+fetched
++++while+x+is+not+None:
++++++++print('got',+x)
++++++++await+sleep(2)
++++++++x+=+await+queue.get()


async+def+main():
++++queue+=+Queue()
++++#+tasks+only+share+the+queue
++++await+gather(
++++++++producer(queue),
++++++++consumer(queue),
++++)


run(main())|code-block|syntax|javascript|1797973|如果您经常需要此功能，还可以将其放入包装异步迭代的helper对象中。助手封装队列和单独的任务。您可以直接在async+for语句中的异步迭代上应用该助手。|1797974|from+asyncio+import+Queue,+sleep,+run,+ensure_future


#+helper+to+consume+iterable+as+concurrent+task
async+def+_enqueue_items(async_iterable,+queue:+Queue,+sentinel):
++++async+for+item+in+async_iterable:
++++++++await+queue.put(item)
++++await+queue.put(sentinel)


async+def+concurrent(async_iterable):
++++"""Concurrently+fetch+items+from+``async_iterable``"""
++++queue+=+Queue()
++++sentinel+=+object()
++++consumer+=+ensure_future(++#+concurrently+fetch+items+for+the+iterable
++++++++_enqueue_items(async_iterable,+queue,+sentinel)
++++)
++++try:
++++++++item+=+await+queue.get()
++++++++while+item+is+not+sentinel:
++++++++++++yield+item
++++++++++++item+=+await+queue.get()
++++finally:
++++++++consumer.cancel()


#+the+original+generator
async+def+g():
++++for+i+in+range(3):
++++++++await+sleep(1)
++++++++yield+i


#+the+original+main+-+modified+with+`concurrent`
async+def+main():
++++async+for+x+in+concurrent(g()):
++++++++print(x)
++++++++await+sleep(2)


run(main())|1797975|entityMap^0|0|5|6|5|Z|M|28|9|0|U|5|0|0|1I|9|0|0^^$0|@$1|2|3|4|5|6|7|S|8|@$9|T|A|U|B|C]|$9|V|A|W|B|C]|$9|X|A|Y|B|C]|$9|Z|A|10|B|C]]|D|@]|E|$]]|$1|F|3|G|5|6|7|11|8|@$9|12|A|13|B|C]]|D|@]|E|$]]|$1|H|3|I|5|J|7|14|8|@]|D|@]|E|$K|L]]|$1|M|3|N|5|6|7|15|8|@$9|16|A|17|B|C]]|D|@]|E|$]]|$1|O|3|P|5|J|7|18|8|@]|D|@]|E|$K|L]]|$1|Q|3|-4|5|6|7|19|8|@]|D|@]|E|$]]]|R|$]]

The point of <code>async</code>/<code>await</code> is to interleave tasks, not functions/generators. For example, when you <code>await asyncio.sleep(1)</code>, your current coroutine is delayed along with the sleep. Similarly, an <code>async for</code> delays its coroutine until the next item is ready.

In order to run your separate functionality, you must create each part as a separate task. Use a <code>Queue</code> to exchange items between them - tasks will only be delayed until they have exchanged an item.

<pre><code>from asyncio import Queue, sleep, run, gather


# the original async generator
async def g():
 for i in range(3):
 await sleep(1)
 yield i


async def producer(queue: Queue):
 async for i in g():
 print('send', i)
 await queue.put(i) # resume once item is fetched
 await queue.put(None)


async def consumer(queue: Queue):
 x = await queue.get() # resume once item is fetched
 while x is not None:
 print('got', x)
 await sleep(2)
 x = await queue.get()


async def main():
 queue = Queue()
 # tasks only share the queue
 await gather(
 producer(queue),
 consumer(queue),
 )


run(main())
</code></pre>

<hr>

If you regularly need this functionality, you can also put it into a helper object that wraps an asynchronous iterable. The helper encapsulates the queue and separate task. You can apply the helper directly on an async iterable in an <code>async for</code> statement.

<pre><code>from asyncio import Queue, sleep, run, ensure_future


# helper to consume iterable as concurrent task
async def _enqueue_items(async_iterable, queue: Queue, sentinel):
 async for item in async_iterable:
 await queue.put(item)
 await queue.put(sentinel)


async def concurrent(async_iterable):
 """Concurrently fetch items from ``async_iterable``"""
 queue = Queue()
 sentinel = object()
 consumer = ensure_future( # concurrently fetch items for the iterable
 _enqueue_items(async_iterable, queue, sentinel)
 )
 try:
 item = await queue.get()
 while item is not sentinel:
 yield item
 item = await queue.get()
 finally:
 consumer.cancel()


# the original generator
async def g():
 for i in range(3):
 await sleep(1)
 yield i


# the original main - modified with `concurrent`
async def main():
 async for x in concurrent(g()):
 print(x)
 await sleep(2)


run(main())
</code></pre>

blocks|key|1626634|text|作为对队列执行此操作的替代方法，此解决方案将期货链连接在一起，以便未来的结果是当前项和检索下一项的另一个未来(可以说，类似于链接列表)：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1626635|from+asyncio+import+sleep,+get_event_loop,+run,+create_task

async+def+aiter(fut,+async_generator):
++++try:
++++++++async+for+item+in+async_generator:
++++++++++++fut,+prev_fut+=+get_event_loop().create_future(),+fut
++++++++++++prev_fut.set_result((item,+fut))
++++++++else:
++++++++++++fut.set_exception(StopAsyncIteration())
++++except+Exception+as+e:
++++++++fut.set_exception(e)


async+def+concurrent(async_generator):
++++fut+=+get_event_loop().create_future()
++++create_task(aiter(fut,+async_generator))

++++try:
++++++++while+True:
++++++++++++item,+fut+=+await+fut
++++++++++++yield+item
++++except+StopAsyncIteration+as+e:
++++++++return|code-block|syntax|javascript|1626636|额外的好处是，该解决方案将正确地处理g()中发生的异常，方法是将main()方法中的异常重命名为具有对调试有用的追溯功能。|1626637|entityMap^0|0|0|0^^$0|@$1|2|3|4|5|6|7|K|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|L|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|M|8|@]|9|@]|A|$]]|$1|I|3|-4|5|6|7|N|8|@]|9|@]|A|$]]]|J|$]]

As an alternative to doing this with a Queue, instead this solution chain Futures together, so that a Future's result is the current item and another Future to retrieve the next item (sort of like a linked list, so to speak):

<pre><code>from asyncio import sleep, get_event_loop, run, create_task

async def aiter(fut, async_generator):
 try:
 async for item in async_generator:
 fut, prev_fut = get_event_loop().create_future(), fut
 prev_fut.set_result((item, fut))
 else:
 fut.set_exception(StopAsyncIteration())
 except Exception as e:
 fut.set_exception(e)


async def concurrent(async_generator):
 fut = get_event_loop().create_future()
 create_task(aiter(fut, async_generator))

 try:
 while True:
 item, fut = await fut
 yield item
 except StopAsyncIteration as e:
 return
</code></pre>

As an added bonus this solution will correctly handle exception that happens in g() by reraising the exception in the main() method with a traceback that will be useful for debugging.

My code is as follows. I want the two sleep can share the same time frame and take 1+2*3=7 seconds to run the script.
But it seems that something wrong happened so that it still takes 3*(1+2) second.

Is there any idea how to modify the code?

<pre class="lang-py prettyprint-override"><code>import asyncio

async def g():
 for i in range(3):
 await asyncio.sleep(1)
 yield i

async def main():
 async for x in g():
 print(x)
 await asyncio.sleep(2)

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

</code></pre>

Python async-generator not async

翻译质量差，导致语言生硬或混乱。

没有提供实际的解决方法或示例。

解答不清晰，无法理解或解决问题。

页面排版不美观，阅读体验差。

文章

问答

视频

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

我的代码如下。我希望这两个睡眠可以共享相同的时间框架，并花费1+2*3=7秒来运行脚本。但是似乎发生了一些错误，所以仍然需要3*(1+2)秒。你知道如何修改代码吗？import asyncioasync def g():    for i in range(3):        await asyncio.sleep(...

问Python异步生成器而不是异步
EN

Stack Overflow用户

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问Python异步生成器而不是异步EN