python - 是否可以在事件循环已在运行时运行 asyncio.Server 实例

Question

我试图了解，是否可以在事件循环已通过 run_forever 方法运行时运行 asyncio.Server 实例(来自单独的线程，当然)。据我了解，服务器可以通过 loop.run_until_complete(asyncio.start_server(...)) 或通过await asyncio.start_server(...)，如果循环已经在运行。第一种方法对我来说是 Not Acceptable ，因为循环已经通过 run_forever 方法运行。但是我也不能使用 await 表达式，因为我将从“循环区域”之外启动它(例如，从不能标记为异步的 main 方法，对吧？)

def loop_thread(loop):
    asyncio.set_event_loop(loop)
    try:
        loop.run_forever()
    finally:
        loop.close()
        print("loop clesed")

class SchedulerTestManager:
    def __init__(self): 
        ...

        self.loop = asyncio.get_event_loop()
        self.servers_loop_thread = threading.Thread(
            target=loop_thread, args=(self.loop, ))
         ...

    def start_test(self):
        self.servers_loop_thread.start()
        return self.servers_loop_thread

    def add_router(self, router):
        r = self.endpoint.add_router(router)
        host = router.ConnectionParameters.Host
        port = router.ConnectionParameters.Port
        srv = TcpServer(host, port)
        server_coro = asyncio.start_server(
            self.handle_connection, self.host, self.port)
        # does not work since add_router is not async
        # self.server = await server_coro
        # does not work, since the loop is already running
        # self.server = self.loop.run_until_complete(server_coro)
        return r


def maind():
   st_manager = SchedulerTestManager()
   thread = st_manager.start_test()
   router = st_manager.add_router(router)

当然，最简单的解决方案是在开始测试(运行循环)之前添加所有路由器(服务器)。但我想尝试实现它，这样就可以在测试已经运行时添加路由器。我认为 loop.call_soon (call_soon_threadsafe) 方法可以帮助我，但似乎不能调度协程，而只是一个简单的函数。

希望我的解释不是很困惑。提前致谢!

Answer 1

为了在一个线程中执行的事件循环与在另一个线程中执行的传统旧线程代码之间进行通信，您可以使用 janus图书馆。

这是一个具有两个接口(interface)的队列:异步接口(interface)和线程安全同步接口(interface)。

这是用法示例:

import asyncio
import janus

loop = asyncio.get_event_loop()
queue = janus.Queue(loop=loop)

def threaded(sync_q):
    for i in range(100):
        sync_q.put(i)
    sync_q.join()

@asyncio.coroutine
def async_coro(async_q):
    for i in range(100):
        val = yield from async_q.get()
        assert val == i
        async_q.task_done()

fut = loop.run_in_executor(None, threaded, queue.sync_q)
loop.run_until_complete(async_coro(queue.async_q))
loop.run_until_complete(fut)

您可以创建一个任务，在循环中等待来自队列的新消息，并根据请求启动新服务器。其他线程可能会将新消息推送到请求新服务器的队列中。