python - 第二次无法执行 celery 节拍

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] Task playstore.tasks.fetch_data_task[3b88c6d0-48db-49c1-b7d1-0b8469775d53]

raised unexpected: TypeError("fetch_data_task() missing 1 required positional argument: 'data'",)

Traceback (most recent call last):

File "/Users/chitrankdixit/.virtualenvs/hashgrowth-> >dev/lib/python3.5/site-packages/celery/app/trace.py", line 240, in >trace_task R = retval = fun(*args, **kwargs) File "/Users/chitrankdixit/.virtualenvs/hashgrowth->dev/lib/python3.5/site-packages/celery/app/trace.py", line 438, in >protected_call return self.run(*args, **kwargs) TypeError: fetch_data_task() missing 1 required positional argument: '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
    },

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