从 celery 任务接收事件

Question

我有一个长时间运行的 celery 任务，它遍历一组项目并执行一些操作。

任务应该以某种方式报告它当前正在处理的项目，以便最终用户知道任务的进度。

目前我的 django 应用程序和 celery 一起在一台服务器上，所以我可以使用 Django 的模型来报告状态，但我计划添加更多远离 Django 的工作人员，所以他们无法访问数据库。

现在我看到几个解决方案：

• 使用一些存储手动存储中间结果，例如 redis 或 mongodb，然后在网络上可用。这让我有点担心，因为例如，如果我将使用 redis，那么我应该在 Django 端保持同步读取状态的代码和写入状态的 Celery 任务，因此它们使用相同的键。
• 使用 REST 调用从 celery 向 Django 报告状态。喜欢PUT http://django.com/api/task/123/items_processed
• 也许使用 Celery 事件系统并创建像 Item processed 这样 django 更新计数器的事件
• 创建一个单独的worker，它在一个带有django的服务器上运行，它持有一个只会增加items proceeded计数的任务，所以当任务完成一个项目时它会发出increase_messages_proceeded_count.delay(task_id)。

我提到的有什么解决办法或隐藏的问题吗？

Answer 1

可能有很多方法可以实现您的目标，但我会这样做。

在您长期运行的 celery 任务中，使用 django's caching framework 设置进度：

from django.core.cache import cache

@app.task()
def long_running_task(self, *args, **kwargs):
    key = "my_task: %s" % self.result.id
    ...
    # do whatever you need to do and set the progress
    # using cache:
    cache.set(key, progress, timeout="whatever works for you")
    ...

然后您所要做的就是使用该键发出一个重复的 AJAX GET 请求并从缓存中检索进度。类似的东西：

 def task_progress_view(request, *args, **kwargs):
     key = request.GET.get('task_key')
     progress = cache.get(key)
     return HttpResponse(content=json.dumps({'progress': progress}),
                         content_type="application/json; charset=utf-8")

这里有一个警告，如果您将服务器作为多个进程运行，请确保您使用的是 memcached 之类的东西，因为 django 的本机缓存将在进程之间不一致。另外，我可能不会使用 celery 的 task_id 作为键，但它足以用于演示目的。

Answer 2

看看flower - 一个用于 Celery 分布式任务队列的实时监控器和网络管理员：

• https://github.com/mher/flower#api
• http://flower.readthedocs.org/en/latest/api.html#get--api-tasks

您需要它来进行演示，对吗？ Flower 与 websocket 一起使用。

例如 - 实时接收任务完成事件（取自官方文档）：

var ws = new WebSocket('ws://localhost:5555/api/task/events/task-succeeded/');
ws.onmessage = function (event) {
    console.log(event.data);
}

您可能需要处理任务 ('ws://localhost:5555/api/tasks/')。

我希望这会有所帮助。

Answer 3

最简单的：

您的任务和 django 应用程序已经共享访问一两个数据存储 - 代理和结果后端（如果您使用的与代理不同）

您可以简单地将一些数据放入这些数据存储中的一个或另一个中，以指示任务当前正在处理的项目。

例如如果使用 redis 只需有一个键“当前正在处理的任务”并将与正在处理的项目相关的数据存储在其中。

Answer 4

您可以使用Swampdragon 之类的东西从 Celery 实例访问用户（您必须能够从客户端访问它，注意不要与 CORS thou 发生冲突）。它可以锁在柜台上，而不是模型本身。

Answer 5

lehins' solution 看起来不错，如果您不介意您的客户反复轮询您的后端。这可能很好，但随着客户数量的增长，它会变得昂贵。

Artur Barseghyan's solution 适用于只需要 Celery 内部机制生成的任务生命周期事件的情况。

或者，您可以使用Django Channels 和 WebSockets 将更新实时推送到客户端。设置非常简单。

1. 将channels 添加到您的INSTALLED_APPS 并设置通道层。例如，使用 Redis 后端：

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [("redis", 6379)]
        }
    }
}

1. 创建事件使用者。这将从 Channels 接收事件并通过 Websockets 将它们推送到客户端。例如：

import json
from asgiref.sync import async_to_sync
from channels.generic.websocket import WebSocketConsumer


class TaskConsumer(WebsocketConsumer):
    def connect(self):
        self.task_id = self.scope['url_route']['kwargs']['task_id'] # your task's identifier
        async_to_sync(self.channel_layer.group_add)(f"tasks-{self.task_id}", self.channel_name)
        self.accept()

    def disconnect(self, code):
        async_to_sync(self.channel_layer.group_discard)(f"tasks-{self.task_id}", self.channel_name)

    def item_processed(self, event):
        item = event['item']
        self.send(text_data=json.dumps(item))

1. 从您的 Celery 任务中推送事件，如下所示：

from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer

...
async_to_sync(get_channel_layer.group_send)(f"tasks-{task.task_id}", {
    'type': 'item_processed',
    'item': item,
})

您还可以编写异步使用者和/或异步调用group_send。无论哪种情况，您都不再需要 async_to_sync 包装器。

4. 将websocket_urlpatterns 添加到您的urls.py：

websocket_urlpatterns = [
    path(r'ws/tasks/<task_id>/', TaskConsumer.as_asgi()),
]

5. 最后，要在客户端使用 JavaScript 中的事件，您可以执行以下操作：

let task_id = 123;
let protocol = location.protocol === 'https:' ? 'wss://' : 'ws://';
let socket = new WebSocket(`${protocol}${window.location.host}/ws/tasks/${task_id}/`);

socket.onmessage = function(event) {
    let data = JSON.parse(event.data);
    let item = data.item;
    // do something with the item (e.g., push it into your state container)
}