1.新進一個django項目

- proj/- proj/__init__.py- proj/settings.py- proj/urls.py - manage.py

2.在該項目創建一個proj / proj / celery.py模塊來定義Celery實例

from __future__ import absolute_importimport osfrom celery import Celery# set the default Django settings module for the 'celery' program. os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings') #注意這里的projfrom django.conf import settings # noqa app = Celery('proj') #創建應用,再這之前要先設置上面的os.environ.setdefault，設置意味著celery命令行程序將知道Django項目的位置 # 你可以在這里直接傳遞對象，但是使用字符串更好，因為當使用Windows或execv時，worker不必序列化對象
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) #打開自動發現apps里面的那些應用有沒有包含tasks.py文件，
#那么Celery應用就會自動去檢索創建的任務。比如你添加了一個任務，在django中會實時地檢索出來。
#例如：這樣您就不必手動將各個模塊添加到CELERY_IMPORTS設置中
#- app1/ # - app1/tasks.py # - app1/models.py #- app2/ # - app2/tasks.py # - app2/models.py
@app.task(bind=True)
def debug_task(self):
　　print('Request: {0!r}'.format(self.request))

?實例2

# -*- coding: utf-8 -*- """ celery 任務示例本地啟動celery命令: python manage.py celery worker --settings=settings 周期性任務還需要啟動celery調度命令：python manage.py celerybeat --settings=settings """ import datetimefrom celery import task from celery.schedules import crontab from celery.task import periodic_taskfrom common.log import logger@task() def async_task(x, y):"""定義一個 celery 異步任務"""logger.error(u"celery 定時任務執行成功，執行結果：{:0>2}:{:0>2}".format(x, y))return x + ydef execute_task():"""執行 celery 異步任務調用celery任務方法:task.delay(arg1, arg2, kwarg1='x', kwarg2='y')task.apply_async(args=[arg1, arg2], kwargs={'kwarg1': 'x', 'kwarg2': 'y'})delay(): 簡便方法，類似調用普通函數apply_async(): 設置celery的額外執行選項時必須使用該方法，如定時（eta）等詳見 ：http://celery.readthedocs.org/en/latest/userguide/calling.html"""now = datetime.datetime.now()logger.error(u"celery 定時任務啟動，將在10s后執行，當前時間：{}".format(now))# 調用定時任務async_task.apply_async(args=[now.hour, now.minute], eta=now + datetime.timedelta(seconds=10))@periodic_task(run_every=crontab(minute='*', hour='*', day_of_week="*")) def get_time():"""celery 周期任務示例run_every=crontab(minute='*/5', hour='*', day_of_week="*")：每 5 分鐘執行一次任務periodic_task：程序運行時自動觸發周期任務"""execute_task()now = datetime.datetime.now()logger.error(u"celery 周期任務調用成功，當前時間：{}".format(now))

3.修改proj / proj /__init__.py

from __future__ import absolute_import #目的是拒絕隱士引入，celery.py和celery沖突。 #這將確保在Django啟動時始終導入應用程序，以便@shared_task裝飾器使用 from .celery import app as celery_app # noqa

4.在配置文件proj/settings.py里面添加

import djcelery djcelery.setup_loader() BROKER_URL = 'redis://127.0.0.1:6379/10'#BROKER_URL：broker是代理人，它負責分發任務給worker去執行。我使用的是Redis作為broker CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/11'#不填就是默認'djcelery.backends.database:DatabaseBackend' # CELERY_RESULT_BACKEND = 'redis://localhost:6379'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai'
CELERY_ENABLE_UTC = False #剛開始時區不準，一直是UTC時間，后來索性把utc禁用
CELERYD_CONCURRENCY = 10
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler' # 定時任務
# 這是使用了django-celery默認的數據庫調度模型,任務執行周期都被存在你指定的orm數據庫中
CELERYD_MAX_TASKS_PER_CHILD = 1 # 每個worker最多執行1個任務就會被銷毀，可防止內存泄露
INSTALLED_APPS += ( 'djcelery')

?實例2

# 是否啟用celery任務 IS_USE_CELERY = True # 本地開發的 celery 的消息隊列（RabbitMQ）信息 BROKER_URL_DEV = 'amqp://guest:guest@127.0.0.1:5672/' # TOCHANGE 調用celery任務的文件路徑, List of modules to import when celery starts. CELERY_IMPORTS = ('home_application.celery_tasks', ) # =============================================================================== # CELERY 配置 # =============================================================================== if IS_USE_CELERY:try:import djceleryINSTALLED_APPS += ('djcelery', # djcelery )djcelery.setup_loader()CELERY_ENABLE_UTC = FalseCELERYBEAT_SCHEDULER = "djcelery.schedulers.DatabaseScheduler"if "celery" in sys.argv:DEBUG = False# celery 的消息隊列（RabbitMQ）信息BROKER_URL = os.environ.get('BK_BROKER_URL', BROKER_URL_DEV)if RUN_MODE == 'DEVELOP':from celery.signals import worker_process_init@worker_process_init.connectdef configure_workers(*args, **kwargs):import djangodjango.setup()except:pass

?

5.創建任務 tasks，在某個應用中demoapp/tasks.py

@task() 裝飾器說明

如果使用了多個裝飾器那么需要task裝飾器在最后即在最上面，一般情況使用的是從celeryapp中引入的app作為的裝飾器：app.task()，如果是django那種在app中定義的task則需要使用@shared_task

from __future__ import absolute_import from celery import shared_task@shared_task def add(x, y):
　　#這里可以把x+y結果寫入數據庫return x + y@shared_task def mul(x, y):return x * y@shared_task def xsum(numbers):return sum(numbers)@task(ignore_result=True,max_retries=1,default_retry_delay=10) #創建測試用的task
def just_print():print "Print from celery task"

6.快速執行

只需要找到方法，用方法加delay就可。

比如上面的文件：

from demoapp.tasks import add r = add.delay(3,5) # 執行這一行就是再下發任務在后臺執行了，結果到時候在取就可以了，
使用技巧，一般會創建一個標識健,存到數據庫，然后傳到我們要執行的方法，在執行方法里面寫上功能，把執行結果保存到指定數據庫，等執行成功后，就用我們之前的標識鍵去查。

7.定時任務的使用（第一種方法-使用數據庫）

使用其實就是修改數據庫，修改后django-celery會實時推送到celery-beat里生效。所以只要再開發一個頁面去配置djcelery_periodictask及其它表就可以了。

?

djcelery提供了一些Model（定義在djcelery/models.py文件），數據庫模型如下，

periodictask	描述定時任務。重要字段有： name: 字符串，標識符 task，字符串，任務函數/類所在的路徑，一般是celery_imports + function name。 interval：外鍵指向intervalschedule，表示每隔多少時間執行 crontab：外鍵指向crontabschedule，表示在某一時刻執行。 enabled：是否生效 expires:? 任務過期時間	?
intervalschedule	表示時間間隔，有兩個參數： every：正數；period間隔單位。比如intervalschedule(every=2, period='day')表示每隔2天	?
crontabschedule	表示某一時刻，有minute、hour、day_of-week、day_of_month、day_of_year它們的組合意義，參見cron時間表示法，比如0 0 * 10 * 表示每個月的10號凌晨。	?

說明：

• 任務和定時任務的區別：定時任務 = 任務 +?intervalschedule/crontabschedule?。兩個定時任務可以執行同一個任務。

• 任務沒有相應的Model，用字符串表示，即periodictask模型的task字段

• 定時任務有相應的Model即periodictask。

• 通過它提供的Model Query API來操作，同平常的數據庫查詢一樣。

from djcelery import models as celery_modelscelery_models.PeriodicTask.objects.create(...) celery_models.PeriodicTask.ojects.get(name='add') ....

8.定時任務示例

admin畢竟是給后臺管理人員使用的，它所有的參數都暴露給使用者了。下面是一個實際使用的例子。

?

需求：ajax實現月度定時任務monthly_reading_task的執行和控制，即

每個月的某一天執行該任務；可以選擇開啟或者關閉該定時任務；能夠選擇任務在哪一天（1-28日）執行。

界面看起來是這樣的：

基本上就是Model的增刪改查。就不是通過admin來操作了。

查詢任務信息

def read(self, request, *args, **kwargs):try:task = celery_models.PeriodicTask.objects.get(name=self.TASK_NAME)if task.enabled:return {'enabled': True,'day_of_month': int(task.crontab.day_of_month),'last_run_at': task.last_run_at if task.last_run_at else '0'}else:return {'enabled': False}except celery_models.PeriodicTask.DoesNotExist:return {'enabled': False}

?

更新日期

def create(self, request, *args, **kwargs):enabled = request.POST.get('enabled', None)if enabled not in [self.ENABLED_POST_VALUE, self.DISABLED_POST_VALUE]:return self.operate_fail('無效參數')if enabled == self.DISABLED_POST_VALUE:self.disable_task(self.TASK_NAME)return self.operate_success()else:try:day_of_month = int(request.POST.get('day_of_month', ''))if day_of_month > 28 or day_of_month < 1:return self.operate_fail('日期必須在1-28日之間')task, created = celery_models.PeriodicTask.objects.get_or_create(name="monthly_reading",task="mrs_app.my_celery.tasks.monthly_reading_task")if created:crontab = celery_models.CrontabSchedule.objects.create(day_of_month=day_of_month,hour=0,minute=0)crontab.save()task.crontab = crontabtask.enabled = Truetask.save()else:task.crontab.day_of_month = day_of_monthtask.crontab.save()task.enabled = Truetask.save()return self.operate_success()except ValueError:return self.operate_fail('抄表日不能為空')

新建定時任務

def celery_get_tag(request):name = 'test'task = 'home_application.celery_tasks.async_task'task_args ={"x":1, "y":1}crontab_time = {'month_of_year':'*','day_of_month':'*','day_of_week':'*','hour':'*','minute':'*'}create_task(name, task , task_args, crontab_time)result = return_result(status=True, code=200, message="添加任務成功")return result#創建任務 def create_task(name, task, task_args, crontab_time):'''name # 任務名字task # 執行的任務 "myapp.tasks.add"task_args # 任務參數 {"x":1, "Y":1}crontab_time # 定時任務時間 格式：{'month_of_year': 9 # 月份'day_of_month': 5 # 日期'hour': 01 # 小時'minute':05 # 分鐘}'''# task任務， created是否定時創建task, created = celery_models.PeriodicTask.objects.get_or_create(name=name,task=task)# 獲取 crontabcrontab = celery_models.CrontabSchedule.objects.filter(**crontab_time).first()if crontab is None:# 如果沒有就創建，有的話就繼續復用之前的crontabcrontab = celery_models.CrontabSchedule.objects.create(**crontab_time)task.crontab = crontab # 設置crontabtask.enabled = True # 開啟tasktask.kwargs = json.dumps(task_args) # 傳入task參數#expiration = timezone.now() + datetime.timedelta(day=1)#task.expires = expiration # 設置任務過期時間為現在時間的一天以后 task.save()result = return_result(status=True, code=200, message="添加任務成功")

關閉定時

def disable_task(self, name):try:task = celery_models.PeriodicTask.objects.get(name=name)task.enabled = Falsetask.save()return Trueexcept celery_models.PeriodicTask.DoesNotExist:return True

9.定時任務的使用（第二種方法-CELERYBEAT_SCHEDULE）

djcelery在初始化中主要完成兩件：

在settings.CELERY_IMPORTS定義下的模塊搜索所有任務。這個對數據庫沒有任何改變，只是用Admin添加定時任務時periodictask.task字段變成選擇框，列出了所有定義的任務。

從settings.CELERYBEAT_SCHEDULE創建定時任務，這個會創建數據記錄，相當于celery_models.PeriodicTask.objects.create(..)語句。

所以在settings文件中可添加配置信息，例如：

CELERYBEAT_SCHEDULE = {'add-every-3-minutes': {'task': 'mrs_app.my_celery.tasks.monthly_reading_task','schedule': timedelta(minutes=3)# 'schedule': crontab(minute=u'40', hour=u'17',), }, }

celerybeat_schedule

定義任務，上面注釋表示每隔3分鐘執行monthly_reading_task任務

??

schedule就是執行計劃，可以用crontab格式，用這種配置，會自動更新數據庫。一般不用這種方法。

?10.啟動

啟動 python manage.py celery worker -l info

如果有定時任務的話，還需要啟動心跳

?另開一個cmd窗口 python manage.py celery beat ?（windows下-B選項不可用）

或者后臺：
/home/python3/bin/python3 /project/manage.py celery worker --loglevel=info >/dev/null 2>&1 &/home/python3/bin/python3 /project/manage.py celery beat >> /project/celery.log 2>&1 &

（參考）官網和https://my.oschina.net/kinegratii/blog/292395

轉載于:https://www.cnblogs.com/CGCong/p/9391436.html

總結

以上是生活随笔為你收集整理的django-celery使用的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。