无法第二次执行 Celery beat

Question

我使用 Celery beat 每 10 秒获取一次站点数据。因此我更新了我的 Django 项目中的设置。我正在使用 rabbitmq 和 celery。

settings.py

# This is the settings file
# Rabbitmq configuration
BROKER_URL = "amqp://abcd:abcd@localhost:5672/abcd"

# Celery configuration
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Kolkata'
CELERY_RESULT_BACKEND = 'djcelery.backends.database:DatabaseBackend'
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'


CELERYBEAT_SCHEDULE = {
    # Executes every Monday morning at 7:30 A.M
    'update-app-data': {
        'task': 'myapp.tasks.fetch_data_task',
        'schedule': timedelta(seconds=10),
        },

celery.py

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

# Indicate Celery to use the default Django settings module
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

app = Celery('myapp')
app.config_from_object('django.conf:settings')
# This line will tell Celery to autodiscover all your tasks.py that are in
# playstore folders
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

app_keywords = Celery('keywords')
app_keywords.config_from_object('django.conf:settings')
# This line will tell Celery to autodiscover all your tasks.py that are in
# keywords folders
app_keywords.autodiscover_tasks(lambda: settings.INSTALLED_APPS)


app1 = Celery('myapp1')
app1.config_from_object('django.conf:settings')
# This line will tell Celery to autodiscover all your tasks.py that are in
# your app folders
app1.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

tasks.py

@task(bind=True)
def fetch_data_task(self, data):
    logger.info("Start task")
    import pdb;pdb.set_trace()
    # post the data to view
    headers, cookies = utils.get_csrf_token()
    requests.post(settings.SITE_VARIABLES['site_url'] + "/site/general_data/",
                  data=json.dumps(data), headers=headers, cookies=cookies
                  )
    if data['reviews']:
        reviews_data = {'app_id': data['app_data'][
            'app_id'], 'reviews': data['reviews'][0]}
        requests.post(settings.SITE_VARIABLES['site_url'] + "/site/blog/reviews/",
                      data=json.dumps(reviews_data), headers=headers, cookies=cookies
                      )
    logger.info("Task fetch data finished")

现在，一旦我在登录网站后在我的 api 中调用 fetch_data_task，该任务就会在 rabbimq 中排队，然后它应该连同参数一起调用该函数。

这是我第一次调用任务的行

tasks.fetch_data_task.apply_async((data,))

这将任务排队并且任务每次都执行，但它给了我以下错误

[2016-09-13 18:57:43,044: ERROR/MainProcess] 任务 playstore.tasks.fetch_data_task[3b88c6d0-48db-49c1-b7d1-0b8469775d53]

引发意外：TypeError("fetch_data_task() missing 1 required positional argument: 'data'",)

Traceback（最近一次调用最后一次）：

文件“/Users/chitrankdixit/.virtualenvs/hashgrowth-> >dev/lib/python3.5/site-packages/celery/app/trace.py”，第 240 行，在 >trace_task R = retval = fun(*args, **kwargs) 文件“/Users/chitrankdixit/.virtualenvs/hashgrowth->dev/lib/python3.5/site-packages/celery/app/trace.py”，第 438 行，在 >protected_call 返回 self.run(*args, **kwargs) TypeError: fetch_data_task() 缺少 1 个必需的位置参数：'data'

如果有人使用过 celery 和 rabbitmq 并且还使用过 celery 处理过周期性任务，请建议我正确执行任务。

Answer 1

异常告诉您错误是什么：您的任务需要位置参数，但您没有在计划定义中提供任何参数。

CELERYBEAT_SCHEDULE = {
    # Executes every Monday morning at 7:30 A.M
    'update-app-data': {
        'task': 'myapp.tasks.fetch_data_task',
        'schedule': timedelta(seconds=10),
        'args': ({
            # whatever goes into 'data' 
        },)  # tuple with one entry, don't omit the comma
    },

从代码中的任何其他位置调用任务不会对计划产生任何影响。