django - 在单独的线程中同步运行异步代码

Question

我正在使用 Django Channels 来支持 websockets，并使用其组的概念向同一组中的多个消费者广播消息。为了向使用者之外发送消息，您需要在同步代码中调用异步方法。不幸的是，这在测试时出现了问题。

我开始使用loop.run_until_complete:

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.ensure_future(channel_layer.group_send(group_name, {'text': json.dumps(message),
                                                                                    'type': 'receive_group_json'}),
                                              loop=loop))

然后堆栈跟踪读取到该线程没有事件循环:RuntimeError: There is no current event loop in thread 'Thread-1'.。为了解决这个问题，我添加了:

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(asyncio.ensure_future(channel_layer.group_send(group_name, {'text': json.dumps(message),
                                                                                    'type': 'receive_group_json'}),
                                              loop=loop))

现在堆栈跟踪正在读取RuntimeError:事件循环已关闭，尽管如果我添加打印语句loop.is_close()打印False.

就上下文而言，我使用 Django 2.0、Channels 2 和 redis 后端。

更新:我尝试在 Python 解释器中运行它(在 py.test 之外以删除移动变量)。当我运行第二个代码块时，我没有收到 Event Loop is close 错误(这可能是由于 Pytest 端的某些原因，无论是超时等)。但是，我在客户端没有收到群消息。然而，我确实看到了一条打印语句:

({<Task finished coro=<RedisChannelLayer.group_send() done, defined at /Users/my/path/to/venv/lib/python3.6/site-packages/channels_redis/core.py:306> result=None>}, set())

更新 2:刷新 redis 后，我在 py.test 中添加了一个固定装置，以刷新每个函数以及 session 范围的事件循环。这次从 RedisChannelLayer 产生另一个打印:

({<Task finished coro=<RedisChannelLayer.group_send() done, defined at /Users/my/path/to/venv/lib/python3.6/site-packages/channels_redis/core.py:306> exception=RuntimeError('Task <Task pending coro=<RedisChannelLayer.group_send() running at /Users/my/path/to/venv/lib/python3.6/site-packages/channels_redis/core.py:316>> got Future <Future pending> attached to a different loop',)>}, set())

Answer 1

如果channel_layer希望驻留在另一个线程中自己的事件循环中，您将需要获取该事件循环对象。一旦你拥有它，你就可以向它提交协程并与你的线程同步，如下所示:

def wait_for_coro(coro, loop):
    # submit coroutine to the event loop in the other thread
    # and wait for it to complete
    future = asyncio.run_coroutine_threadsafe(coro, loop)
    return future.wait()

wait_for_coro(channel_layer.group_send(group_name, ...), channel_loop)