异步发电机

Abhinavkulkarni

我有以下情况：

我有一个阻塞的同步发电机
我有一个非阻塞的异步功能

我想运行事件生成器上的阻塞生成器（在中执行ThreadPool）和async函数。我该如何实现？

以下函数只是打印生成器的输出，而不是sleep函数的输出。

谢谢！

from concurrent.futures import ThreadPoolExecutor

import numpy as np
import asyncio
import time


def f():
    while True:
        r = np.random.randint(0, 3)
        time.sleep(r)
        yield r


async def gen():
    loop = asyncio.get_event_loop()
    executor = ThreadPoolExecutor()
    gen = await loop.run_in_executor(executor, f)
    for item in gen:
        print(item)
        print('Inside generator')


async def sleep():
    while True:
        await asyncio.sleep(1)
        print('Inside async sleep')


async def combine():
    await asyncio.gather(sleep(), gen())


def main():
    loop = asyncio.get_event_loop()
    loop.run_until_complete(combine())


if __name__ == '__main__':
    main()

用户名

run_in_executor在生成器上不起作用，因为它是为阻止功能而设计的。尽管生成器是有效函数，但生成器在被调用时会立即返回，从而提供调用者应该通过重复调用来耗尽的对象next。（这是Pythonfor循环在后台执行的操作。）要使用异步代码的阻塞生成器，您有两种选择：

将迭代的每个步骤（对的每个单独调用next）包装到对run_in_executor或的单独调用中
for在单独的线程中启动循环，然后使用队列将对象传输到异步使用者。

两种方法都可以抽象为一个接受迭代器并返回等效异步迭代器的函数。这是第二种方法的实现：

import asyncio, threading

def async_wrap_iter(it):
    """Wrap blocking iterator into an asynchronous one"""
    loop = asyncio.get_event_loop()
    q = asyncio.Queue(1)
    exception = None
    _END = object()

    async def yield_queue_items():
        while True:
            next_item = await q.get()
            if next_item is _END:
                break
            yield next_item
        if exception is not None:
            # the iterator has raised, propagate the exception
            raise exception

    def iter_to_queue():
        nonlocal exception
        try:
            for item in it:
                # This runs outside the event loop thread, so we
                # must use thread-safe API to talk to the queue.
                asyncio.run_coroutine_threadsafe(q.put(item), loop).result()
        except Exception as e:
            exception = e
        finally:
            asyncio.run_coroutine_threadsafe(q.put(_END), loop).result()

    threading.Thread(target=iter_to_queue).start()
    return yield_queue_items()

可以使用一个普通的time.time()用于阻止的同步迭代器和一个异步心跳函数来测试事件循环，以对其进行测试：

# async_wrap_iter definition as above

import time

def test_iter():
    for i in range(5):
        yield i
        time.sleep(1)

async def test():
    ait = async_wrap_iter(test_iter())
    async for i in ait:
        print(i)

async def heartbeat():
    while True:
        print('alive')
        await asyncio.sleep(.1)

async def main():
    asyncio.create_task(heartbeat())
    await test()

asyncio.run(main())

本文收集自互联网，转载请注明来源。

如有侵权，请联系 [email protected] 删除。