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

Question

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

我有一个 Django 项目，它的主要目的是从 DB rapidly 执行排队的任务。我使用 Celery 和 Celerybeat 通过 Django 通道实现这一点，以实时更新我的模板响应。

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

我的任务（简化版）：

@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] actions queued: 25

[2019-10-03 18:48:02,206: WARNING/MainProcess] c:\users\jack\documents\github\mcr-admin\venv\lib\site-packages\gevent_socket3.py:123: RuntimeWarning: coroutine 'AsyncToSync.main_wrap' was never awaited
self._read_event = io_class(fileno, 1)

RuntimeWarning: Enable tracemalloc to get the object allocation traceback

[2019-10-03 18:48:02,212: WARNING/MainProcess] c:\users\jack\documents\github\mcr-admin\venv\lib\site-packages\gevent_socket3.py:123: RuntimeWarning: coroutine 'BaseEventLoop.shutdown_asyncgens' was never awaited 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 非常靠近，不断地在 link 中抛出错误。

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

这里的问题是，有时在我添加的协程有机会完成之前，asyncio 事件循环会停止，我一直在调整我的代码，尝试使用 asyncio 和事件循环 api 但是我要么破坏我的代码，要么得到更糟糕的结果。我觉得这可能与 Asgiref async_to_sync 函数有关，但它很复杂，我才开始使用 python异步几天前。

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

干杯。

Answer 1

您好，我目前正遇到您的确切问题，即能够将已完成的芹菜任务中的消息发送到客户端至关重要。

我之前能够 group_send 通过对模型方法使用信号，例如：

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'
   }))

Answer 2

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

消费者：

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

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

Django Channels send group message from Celery task. Asyncio event loop stopping before all async tasks finished

django

celery

gevent

channels

python-3.7