时间戳

首页 > 解决方案 > 收到芹菜节拍任务但未调用

python - 收到芹菜节拍任务但未调用

问题描述

我正在尝试运行节拍任务。但只执行一项任务。所以，有我的项目结构：

交易平台：

芹菜.py：

import os
from celery import Celery
from django.conf import settings

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'trading_platform.settings')
app = Celery('trading_platform')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)


@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    sender.add_periodic_task(10.0, debug_task.s('HELLO'), name='add every 10')


@app.task
def debug_task(self):
    print(self)

设置.py：

# Celery Configuration Options
CELERY_BROKER_URL = os.environ.get('CELERY_BROKER_URL', 'redis://redis:6379/0')
CELERY_RESULT_BACKEND = os.environ.get('CELERY_RESULT_BACKEND', 'redis://redis:6379/0')
# CELERY_ACCEPT_CONTENT = os.environ.get('CELERY_ACCEPT_CONTENT', 'application/json')
CELERY_RESULT_SERIALIZER = os.environ.get('CELERY_RESULT_SERIALIZER', 'json')
CELERY_TASK_SERIALIZER = os.environ.get('CELERY_TASK_SERIALIZER', 'json')
CELERY_STORE_ERRORS_EVEN_IF_IGNORED = os.environ.get('CELERY_STORE_ERRORS_EVEN_IF_IGNORED', True)

任务.py：

from trading_platform.celery import app


@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    sender.add_periodic_task(10.0, print_text.s('text'), name='add every 10 sec')


@app.task(bind=True)
def print_text(text):
    print(text)

运行命令：celery -A trading_platform worker -B -l info

所以，有输入：

celery_1  | /usr/local/lib/python3.8/site-packages/celery/platforms.py:797: RuntimeWarning: You're running the worker with superuser privileges: this is
celery_1  | absolutely not recommended!
celery_1  | 
celery_1  | Please specify a different user using the --uid option.
celery_1  | 
celery_1  | User information: uid=0 euid=0 gid=0 egid=0
celery_1  | 
celery_1  |   warnings.warn(RuntimeWarning(ROOT_DISCOURAGED.format(
celery_1  |  
celery_1  |  -------------- celery@953a7a853036 v5.0.1 (singularity)
celery_1  | --- ***** ----- 
celery_1  | -- ******* ---- Linux-5.4.0-52-generic-x86_64-with-glibc2.2.5 2020-10-25 09:23:38
celery_1  | - *** --- * --- 
celery_1  | - ** ---------- [config]
celery_1  | - ** ---------- .> app:         trading_platform:0x7f2b34bc2520
celery_1  | - ** ---------- .> transport:   redis://redis:6379/0
celery_1  | - ** ---------- .> results:     redis://redis:6379/0
celery_1  | - *** --- * --- .> concurrency: 12 (prefork)
celery_1  | -- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
celery_1  | --- ***** ----- 
celery_1  |  -------------- [queues]
celery_1  |                 .> celery           exchange=celery(direct) key=celery
celery_1  |                 
celery_1  | 
celery_1  | [tasks]
celery_1  |   . offers.tasks.print_text
celery_1  |   . trading_platform.celery.debug_task
celery_1  | 
celery_1  | [2020-10-25 09:23:39,191: INFO/MainProcess] Connected to redis://redis:6379/0
celery_1  | [2020-10-25 09:23:39,197: INFO/MainProcess] mingle: searching for neighbors

celery_1  | [2020-10-25 09:24:20,632: INFO/Beat] Scheduler: Sending due task add every 10 (trading_platform.celery.debug_task)
celery_1  | [2020-10-25 09:24:20,638: INFO/MainProcess] Received task: trading_platform.celery.debug_task[c384669c-951c-4b67-befd-f4683f048ca2]  
celery_1  | [2020-10-25 09:24:20,641: WARNING/ForkPoolWorker-8] HELLO
celery_1  | [2020-10-25 09:24:20,643: INFO/ForkPoolWorker-8] Task trading_platform.celery.debug_task[c384669c-951c-4b67-befd-f4683f048ca2] succeeded in 0.002888333000100829s: None

那么，我们如何才能看到只执行了一项任务（来自 trading_platform/celery.py）。如何解决？

标签： pythondjangocelery

解决方案

除了正常的 celery worker 之外，还必须启动一个 celery beat 过程。还建议不要使用该-B选项。相反，您应该分别启动 worker 和 beat scheduler。

此外，对于 django，您必须将调度程序指定为django_celery_beat.schedulers:DatabaseScheduler通过django-celery-beat

将 django-celery-beat 添加到您的 django 应用程序，如果您还没有：

# settings.py
INSTALLED_APPS = (
    ...,
    'django_celery_beat',
)

启动节拍调度程序：

$ celery -A proj beat -l INFO --scheduler django_celery_beat.schedulers:DatabaseScheduler

启动工人：

$ celery -A proj worker -l INFO

参考：https ://docs.celeryproject.org/en/stable/userguide/periodic-tasks.html#starting-the-scheduler

此外，由于您bind=True为任务指定，您需要修改签名以使其接受任务对象。

@app.task(bind=True)
def print_text(self, text):
    print(text)

这里，self是任务对象。

推荐阅读