Celery 实时监控和 Django 应用

Question

Celery docs 解释如何设置自定义事件处理。然而，这种方法似乎不太适用于我的 Django 项目和 celery。

我的celery.py 文件看起来像

from __future__ import absolute_import 
import os
from celery import Celery
from django.conf import settings
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproj.settings')

# The custom monitor copied from celery docs
def my_monitor(app):
    state = app.events.State()

    def announce_failed_tasks(event):
        state.event(event)
        # task name is sent only with -received event, and state
        # will keep track of this for us.
        task = state.tasks.get(event['uuid'])

        print('MY MON TASK FAILED: %s[%s] %s' % (
            task.name, task.uuid, task.info(), ))

    with app.connection() as connection:
        recv = app.events.Receiver(connection,
                                   handlers={'task-failed': announce_failed_tasks, }
                                   )
        recv.capture(limit=None, timeout=None, wakeup=True)


app = Celery('myproj')


app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
# Monitoring
my_monitor(app)

我刚刚从文档中复制了示例监视器。

我没有收到错误，但启动 celery worker/beat 或 django 的 runserver 似乎永远挂起，直到我注释掉 my_monitor(app) 行。

在 Django 项目中执行此操作的正确方法是什么？

Answer 1

recv.capture 是一个阻塞调用。您是否从事件处理程序中获得任何输出？

如果您尝试将您的 django 服务器用作监视器和任务生产者/消费者，那是行不通的。

您的显示器应该是一个独立的独立应用程序。您还应该考虑消除 django 依赖项，因为它是不必要的。

Answer 2

我建议使用 celery-flower 来执行此操作。

Flower 启动一个 tornado WebSocket 服务器并将任务的结果和状态传送到 API WebSocket 端点。

请在下面找到一个 Django 事件模板的示例

{% extends "admin/base.html" %}
{% load staticfiles %}

{% block extrastyle %}
<style>
    table, th, td {
    border: 1px solid black;
        border-collapse: collapse;
    }
    th, td {
        padding: 5px;
    }
</style>
{% endblock extrastyle %}

{% block extrahead %}
<title>Read Flower event Websocket-API</title>
<script language="javascript" type="text/javascript">
    var ws_success = new WebSocket('ws://localhost:5555/api/task/events/task-succeeded/');
    ws_success.onmessage = function (event) {
        var div = document.getElementById("output");
        var li = document.createElement('li');
        li.innerText = event.data;
        div.appendChild(li);
    }
</script>
{% endblock extrahead %}

{% block content %}
    <h2>Events:</h2>
    <ul>
    <div id="output"></div>
    </ul>
{% endblock content %}

django 模板使用以下端点：

ws://localhost:5555/api/task/events/task-succeeded/

是否可以使用第二个 WebSocket 端点：

ws://localhost:5555/dashboard

用“pipenv”安装它（比pip and virtualenv更可取）：

pipenv install flower

开始花：

flower -A $projectname -l info

和芹菜工人与花的事件信息：

celery -A $projectname worker -E -B -l info

见celery heartbeat 和celery command line options for heartbeat and events。 Celery beat period 是任务调度程序，将使用“-E”选项启动。

在以下文件中配置 celery 后，启动你的 Django 项目：

• $projectname/celery.py
• $projectname/init.py
• celeryconfig.py

如celery configuration 中所述。