Django Channels 从 Celery 任务发送组消息。 Asyncio 事件循环在所有异步任务完成之前停止

Question

我目前遇到一个特别棘手的问题，我会尽力解释它。

我有一个 Django 项目，它的主要目的是快速执行数据库中的排队任务。我使用 Celery 和 Celerybeat 通过 Django 通道来实现这一点，以实时更新我的​​模板和响应。

Celery worker 是一个 gevent 工作池，具有相当数量的线程。

我的任务（简化版）：

@shared_task
def exec_task(action_id):
  # execute the action
  action = Action.objects.get(pk=action_id)
  response = post_request(action)

  # update action status
  if response.status_code == 200:
    action.status = 'completed'

  else:
    action.status = 'failed'

  # save the action to the DB
  action.save()

  channel_layer = get_channel_layer()
  status_data = {'id': action.id, 'status': action.status}
  status_data = json.dumps(status_data)
  try:
    async_to_sync(channel_layer.group_send)('channel_group', {'type': 'propergate_status', 'data': status_data})
  except:
    event_loop = asyncio.get_running_loop()
    future = asyncio.run_coroutine_threadsafe(channel_layer.group_send('channel_group', {'type': 'propergate_status', 'data': status_data}), event_loop)
    result = future.result()

我的错误：

[2019-10-03 18:47:59,990: WARNING/MainProcess] 动作排队：25

[2019-10-03 18:48:02,206：警告/MainProcess] c:\users\jack\documents\github\mcr-admin\venv\lib\site-packages\gevent_socket3.py:123： RuntimeWarning：从未等待协程“AsyncToSync.main_wrap”
self._read_event = io_class(fileno, 1)

RuntimeWarning: E​​nable tracemalloc 获取对象分配回溯

[2019-10-03 18:48:02,212：警告/MainProcess] c:\users\jack\documents\github\mcr-admin\venv\lib\site-packages\gevent_socket3.py:123： RuntimeWarning：协程“BaseEventLoop.shutdown_asyncgens”从来没有 等待 self._read_event = io_class(fileno, 1) RuntimeWarning：

最初在我将操作保存到我刚刚调用的数据库之后：

async_to_sync(channel_layer.group_send)('channel_group', {'type': 'propergate_status', 'data': status_data})

但我一直收到运行时错误，因为如果已经有一个 asyncio 事件循环已经在运行，您就不能使用 async_to_sync，as shown here at line 61。所以我有多个 gevent 线程试图 async_to_sync 非常靠近，不断地在链接中抛出错误。

这让我找到了this wonderful answer 和当前版本的 exec_task，它在向 Django Channels 组发送消息时的成功率为 98%，但我真的需要它是 100%。

这里的问题是，在我添加的协程有机会完成之前，偶尔会停止 asyncio 事件循环，并且我一直在调整我的代码，使用 asyncio 和事件循环 api，但我要么破坏我的代码，要么得到更差的结果。我感觉这可能与 Asgiref async_to_sync 函数提前关闭循环有关，但这很复杂，我几天前才开始使用 python async。

欢迎任何反馈、cmets、提示或修复！

干杯。

Answer 1

最后我无法解决问题并选择使用 Channels AsyncHttpConsumer 发送群组消息的替代解决方案。它不是最佳的，但它可以工作并将工作流保留在 Channels 库中。

消费者：

class celeryMessageConsumer(AsyncHttpConsumer):

async def handle(self, body):
    # send response
    await self.send_response(200, b"Recieved Loud and Clear", headers=[
        (b"Content-Type", b"text/plain"),
    ])
    # formating url encoded string into json
    body_data = urllib.parse.unquote_plus(body.decode("utf-8"))
    body_data = json.loads(body_data)
    id = body_data['data']['id']

    await self.channel_layer.group_send(
        f"group_{id}",
        {
            'type': 'propergate.data',
            'data': body_data['data']
        }
    )

路由：

application = ProtocolTypeRouter({
    'websocket': AuthMiddlewareStack(
        URLRouter(
            myApp.routing.websocket_urlpatterns
        )
    ),
    'http': URLRouter([
        path("celeryToTemplate/", consumers.celeryMessageConsumer),
        re_path('genericMyAppPath/.*', AsgiHandler),
    ]),
})

Http 请求：

data = json.dumps({'id': id, 'status': status})
response = internal_post_request('http://genericAddress/celeryToTemplate/', data)
if response.status_code == 200:
    # phew
    pass
else:
    # whoops
    pass

请求：

def internal_post_request(request_url, payload):
    headers={
        'Content-Type': 'application/json'
    }
    response = requests.post(request_url, data=payload, headers=headers)
    return response

Answer 2

您好，我目前遇到了您的确切问题，即能够从已完成的 celery 任务向客户端发送消息至关重要。

我之前可以通过使用信号到模型方法来分组发送，例如：

def SyncLogger(**kwargs):
    """ a syncronous function to instigate the websocket layer
    to send messages to all clients in the project """

instance = kwargs.get('instance')
# print('instance {}'.format(instance))

args = eval(instance.args)
channel_layer = channels.layers.get_channel_layer()
async_to_sync(channel_layer.group_send)(
    args ['room'],
    {
        "type": "chat.message",
        "operation": args['operation'],
        "state": instance.state,
        "task": instance.task
    })

和信号

post_save.connect(SyncLogger, TaskProgress)

更新 只要有event_loop，我就可以发送消息 无论消费者是否异步，这都有效

@shared_task()
def test_message():
   channel_layer = get_channel_layer()

   loop = asyncio.new_event_loop()
   asyncio.set_event_loop(loop)

   loop.run_until_complete(channel_layer.group_send('sync_chat', {
       'type': 'chat.message',
       'operation': 'operation',
       'state': 'state',
       'task': 'task'
   }))