在异步 django 通道 websocket 中使用同步 celery 回调答案

【问题标题】：Using a synchronous celery callback in an async django channels websocket在异步 django 通道 websocket 中使用同步 celery 回调
【发布时间】：2019-11-18 21:46:38
【问题描述】：

在我们的 django-channels 异步消费者中，我正在尝试检索组结果。这个结果可能是也可能不是完整的，因此我想在管道结果准备好时添加一个回调。问题是，websocket 是异步的，回调必须是同步的，self.send 方法必须是异步的。最初我假设能够将异步发送包装在async_to_sync 中，尽管它给出了错误You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly.，因此似乎不允许异步->同步->异步。

class ImageConsumer(AsyncWebsocketConsumer):

    async def celery_retrieve_group_result(self, batch_id):
        group = group = celery.GroupResult().restore(batch_id)
        res = group.get(callback=self.sync_callback)

    def sync_callback(self, task_id, value):
        async_to_sync(self.send_group_result)(task_id, value)

    async def send_group_result(self, task_id , value):
        await self.send(json.dumps({"type": "result.redraw_border", "result": value}))

我尝试了不同的组合，但我的限制因素是回调必须是同步的，而其他一切都是异步的。有没有人以这种方式将芹菜与异步 django 混合的经验？任何帮助将不胜感激！

【问题讨论】：

标签： asynchronous websocket celery django-channels python-asyncio

【解决方案1】：

决定采用不同的方法来迭代封闭的 AsyncResults。

class ImageConsumer(AsyncWebsocketConsumer):
    name = "Image Consumer"
    timeout = 20
    sleep_duration = 5e-2

    async def celery_retrieve_group_result(self, batch_id):

        group = celery.GroupResult().restore(batch_id)
        if group is not None:
            ordered_list_of_async_results = group.results
            for result in ordered_list_of_async_results:
                start = time.time()
                try:
                    while not result.ready():
                        await asyncio.sleep(self.sleep_duration)
                        if self.timeout and time.time() - start > self.timeout:
                            logger.exception(str(TimeoutError))
                            raise TimeoutError
                    value = result.get()
                    await self.send(json.dumps({'type': 'result.processing', 'result': value}))
                except:
                    await self.send(json.dumps({'type': 'result.processing', 'error': str(TimeoutError)}))
            group.forget()
        else:
            await self.send(json.dumps({'type': 'response.nonexistant_group', 'error': 'No result batch exists. re-upload'}))

这真的很好用。超时是因为 Celery 4.3 当前存在 Redis 后端问题，如果删除该问题会导致死机。因此，在解决此问题之前，这是完美的。

【讨论】：