如何将 Python 生成器转换为异步生成器？

是的，不幸的是，这并不简单。

下面的代码段中的可选内容是“任务组”，但它有助于良好实践：

import time

import asyncio
from concurrent.futures import ThreadPoolExecutor # (or Process*)

executor = ThreadPoolExecutor(10)  # Put this wherever you like, with as many workers as you want.

def blocking_task(i):
    time.sleep(1)
    return i

async def async_gen():
    """ plain generator with blocking i/o """
    async with asyncio.TaskGroup() as tg:
        loop = asyncio.get_running_loop()
        async def taskwrapper(awaitable):
            return await awaitable
        tasks = {tg.create_task(taskwrapper(loop.run_in_executor(executor, blocking_task, i,))) for i in range(10)}
        while tasks:
            done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
            for task in done:
                yield task.result()
            tasks = pending


async def consumer():
    async for i in async_gen():
        print(i)


asyncio.run(consumer())

因此，这时您可以在一个任务中分离每个项目提取，该任务可以在线程池中作为单独的任务运行。

如果你只想保留一个同步流，然后包装它，以便它可以在异步循环中使用，如果你可以在它自己的线程中运行该同步生成器，这可以通过添加一些辅助函数和控制流的东西来完成 - 下面的代码片段可能会有所帮助。该函数将负责在另一个线程中启动生成器并异步获取块：async_gen

import asyncio
from threading import Thread
from queue import Queue as ThreadingQueue, Empty


import time


def blocking():
    """ plain generator with blocking i/o """
    # this will run in other thread, and produced results
    # are comunicated to the asyncio loop in a queue by the
    # "middleware" helper functions
    i = 0
    while True:
        time.sleep(1)
        yield i
        i += 1


def _threading_queue_item_ready(semaphore):
    # increase the count of available items in the queue so
    # that it can be read in an async-friendly way:
    semaphore.release()

def _producer_wrapper(gen, loop, queue, semaphore):
    def wrapper(*args, **kwargs):
        for item in gen():
            queue.put(item)
            loop.call_soon_threadsafe(_threading_queue_item_ready, semaphore)
    wrapper()

async def async_gen(sync_gen):
    """wrapper for the "generator asynchronization" mechanism"""
    semaphore = asyncio.Semaphore(0)
    queue = ThreadingQueue()
    producer_thread = Thread(target=_producer_wrapper, args=(sync_gen, asyncio.get_running_loop(), queue, semaphore))
    producer_thread.start()
    while True:
        await semaphore.acquire()
        # we only get here when there is at least one item in the queue
        item = queue.get()
        yield item
        # intentionaly do not "release" the semaphore: available item count
        # is increased by the producer code

async def consumer():
    async for chunk in async_gen(blocking):
        print(chunk)


asyncio.run(consumer())