芹菜和Django的定期任务

我遇到了一些问题，需要定期运行Celery 3.1.8，Django 1.6.1和RabbitMQ。我对目前的文档有些困惑，因为据我所知django-celery不再需要Celery和Django一起运行。我有一种感觉，我没有正确地运行工作人员，但是在搜索SO和Google搜索解决方案之后，我需要帮助。任何人都可以用这个指向正确的方向吗？芹菜和Django的定期任务

settings.py（不知道我需要这个，因为我有我的任务@periodic_task装饰）

CELERYBEAT_SCHEDULE = { 
    'add-every-30-seconds': { 
     'task': 'tasks.send_test_email', 
     'schedule': datetime.timedelta(seconds=30) 
    }, 
}

我的应用程序（celery.py）

from __future__ import absolute_import 

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

# set the default Django settings module for the 'celery' program. 
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'app.settings') 

app = Celery('app', 
      broker='amqp://', 
      backend='amqp://', 
      include=['app.tasks']) 


app.conf.update(
    CELERY_TASK_RESULT_EXPIRES=3600, 
    CELERY_TIMEZONE='Europe/Oslo', 
    ) 

if __name__ == '__main__': 
    app.start() 

# Using a string here means the worker will not have to 
# pickle the object when using Windows. 
app.config_from_object('django.conf:settings') 
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) 


@app.task(bind=True) 
def debug_task(self): 
    print('Request: {0!r}'.format(self.request))

周期性任务（任务的.py）

from __future__ import absolute_import 
from celery.task import periodic_task 
import datetime 

@periodic_task(run_every=datetime.timedelta(minutes=1)) 
def send_test_email(): 
    print "This is a periodic task from celery"

在命令行中，我执行了工人：

celery worker -A app -l info 
celery beat

来源

2014-01-30 Cliff F

所以修修补补各地的之后，我能得到它的工作。问题在于我如何执行工作人员。这样做的诀窍：**芹菜 - 一个应用程序击败** –

你应该添加，作为一个答案，而不是评论:) – Jonathan

一些其他的解决办法是使用@periodic_taskcelery decorator

from celery.schedules import crontab 

@periodic_task(run_every=crontab(minute=0, hour=1)) 
def my_task(): 
    print 'my_task'

来源

2015-01-20 17:43:42

芹菜和Django的定期任务

回答

相关问题