提交作业到异步事件循环 [英] Submit a job to an asyncio event loop

问题描述

我想将作业从线程提交到asyncio事件循环(就像

I would like to submit jobs from a thread to an asyncio event loop (just like run_in_executor but the other way around).

这是asyncio文档关于并发的内容和多线程:

要安排从其他线程进行的回调，应使用BaseEventLoop.call_soon_threadsafe()方法. 从其他线程安排协程的示例: loop.call_soon_threadsafe(asyncio.async, coro_func())

To schedule a callback from a different thread, the BaseEventLoop.call_soon_threadsafe() method should be used. Example to schedule a coroutine from a different thread: loop.call_soon_threadsafe(asyncio.async, coro_func())

这很好，但是协程的结果丢失了.

That works fine but the result of the coroutine is lost.

相反，可以使用向async(或ensure_future)返回的将来添加完成的回调的函数，以便线程可以通过

Instead, it is possible to use a function that adds a done callback to the future returned by async (or ensure_future) so that the thread can access the result through a concurrent.futures.Future.

为什么没有在标准库中实现这种功能的特殊原因?还是我错过了一种更简单的方法来实现这一目标?

Is there a particular reason why such a feature is not implemented in the standard library? Or did I miss a simpler way to achieve that?

推荐答案

我的请求成功了，并且

My request made its way and a run_coroutine_threadsafe function has been implemented here.

示例:

def target(loop, timeout=None):
    future = asyncio.run_coroutine_threadsafe(add(1, b=2), loop)
    return future.result(timeout)

async def add(a, b):
    await asyncio.sleep(1)
    return a + b

loop = asyncio.get_event_loop()
future = loop.run_in_executor(None, target, loop)
assert loop.run_until_complete(future) == 3

---

我最初发布了 concurrent的子类.futures.Executor 仍可以实现为:

---

I originally posted a sub-class of concurrent.futures.Executor that can still be implemented as:

class LoopExecutor(concurrent.futures.Executor):
    """An Executor subclass that uses an event loop 
    to execute calls asynchronously."""

    def __init__(self, loop=None):
        """Initialize the executor with a given loop."""
        self.loop = loop or asyncio.get_event_loop()

    def submit(self, fn, *args, **kwargs):
        """Schedule the callable, fn, to be executed as fn(*args **kwargs).
        Return a Future object representing the execution of the callable."""
        coro = asyncio.coroutine(fn)(*args, **kwargs)
        return asyncio.run_coroutine_threadsafe(coro, self.loop)

这篇关于提交作业到异步事件循环的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！