pip3 install django #當(dāng)前我安裝的版本是 3.0.6

django-admin startproject typeidea
django-admin startapp blog

pip3 install django-celery
pip3 install -U Celery 
pip3 install "celery[librabbitmq,redis,auth,msgpack]" 
pip3 install django-celery-beat # 用于動(dòng)態(tài)添加定時(shí)任務(wù)
pip3 install django-celery-results
pip3 install redis

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
 
"""
#File: tasks.py
#Time: 2022/3/30 2:26 下午
#Author: julius
"""
from celery import Celery
 
# 使用redis做為broker
app = Celery('blog.tasks2',broker='redis://127.0.0.1:6379/0')
 
# 創(chuàng)建任務(wù)函數(shù)
@app.task
def my_task():
    print('任務(wù)正在執(zhí)行...')

$ python manage.py shell
>>> from blog.tasks import my_task
>>> my_task.delay()
<AsyncResult: 83484dfe-f729-417b-8e51-6c7ae32a1377>

$ python manage.py shell
>>> from blog.tasks import my_task
>>> ret=my_task.delay()
>>> ret.result()

from celery import Celery
 
# 使用redis作為broker以及backend
app = Celery('celery_tasks.tasks',
             broker='redis://127.0.0.1:6379/8',
             backend='redis://127.0.0.1:6379/9')
 
# 創(chuàng)建任務(wù)函數(shù)
@app.task
def my_task(a, b):
    print("任務(wù)函數(shù)正在執(zhí)行....")
    return a + b

$ python manage.py shell
>>> from blog.tasks import my_task
>>> res=my_task.delay(10,40)
>>> res.result
50
>>> res.failed()
False

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
 
"""
#File: celery.py
#Time: 2022/3/30 12:25 下午
#Author: julius
"""
import os
from celery import Celery
from blog import celeryconfig
project_name='typeidea'
# set the default django setting module for the 'celery' program
os.environ.setdefault('DJANGO_SETTINGS_MODULE','typeidea.settings')
app = Celery(project_name)
 
app.config_from_object('django.conf:settings')
 
app.autodiscover_tasks()

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
 
"""
#File: celeryconfig.py
#Time: 2022/3/30 2:54 下午
#Author: julius
"""

# 設(shè)置結(jié)果存儲(chǔ)
from typeidea import settings
import os
 
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "typeidea.settings")
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'
# 設(shè)置代理人broker
BROKER_URL = 'redis://127.0.0.1:6379/1'
# celery 的啟動(dòng)工作數(shù)量設(shè)置
CELERY_WORKER_CONCURRENCY = 20
# 任務(wù)預(yù)取功能，就是每個(gè)工作的進(jìn)程／線程在獲取任務(wù)的時(shí)候，會(huì)盡量多拿 n 個(gè)，以保證獲取的通訊成本可以壓縮。
CELERYD_PREFETCH_MULTIPLIER = 20
# 非常重要,有些情況下可以防止死鎖
CELERYD_FORCE_EXECV = True
# celery 的 worker 執(zhí)行多少個(gè)任務(wù)后進(jìn)行重啟操作
CELERY_WORKER_MAX_TASKS_PER_CHILD = 100
# 禁用所有速度限制，如果網(wǎng)絡(luò)資源有限，不建議開(kāi)足馬力。
CELERY_DISABLE_RATE_LIMITS = True
 
CELERY_ENABLE_UTC = False
CELERY_TIMEZONE = settings.TIME_ZONE
DJANGO_CELERY_BEAT_TZ_AWARE = False
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

import time
from blog.celery import app
 
# 創(chuàng)建任務(wù)函數(shù)
@app.task
def my_task(a, b, c):
    print('任務(wù)正在執(zhí)行...')
    print('任務(wù)1函數(shù)休眠10s')
    time.sleep(10)
    return a + b + c

pip3 install django-celery-beat

INSTALLED_APPS = [
    'blog',
    'django_celery_beat',
    ...
]
 
# Django設(shè)置時(shí)區(qū)
LANGUAGE_CODE = 'zh-hans'  # 使用中國(guó)語(yǔ)言
TIME_ZONE = 'Asia/Shanghai'  # 設(shè)置Django使用中國(guó)上海時(shí)間
# 如果USE_TZ設(shè)置為True時(shí)，Django會(huì)使用系統(tǒng)默認(rèn)設(shè)置的時(shí)區(qū)，此時(shí)的TIME_ZONE不管有沒(méi)有設(shè)置都不起作用
# 如果USE_TZ 設(shè)置為False,TIME_ZONE = 'Asia/Shanghai', 則使用上海的UTC時(shí)間。
USE_TZ = False

import os
 
from celery import Celery
 
from blog import celeryconfig
 
# 為celery 設(shè)置環(huán)境變量
os.environ.setdefault("DJANGO_SETTINGS_MODULE","typeidea.settings")
# 創(chuàng)建celery app
app = Celery('blog')
# 從單獨(dú)的配置模塊中加載配置
app.config_from_object(celeryconfig)
 
# 設(shè)置app自動(dòng)加載任務(wù)
app.autodiscover_tasks([
    'blog',
])

# 設(shè)置結(jié)果存儲(chǔ)
from typeidea import settings
import os
 
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "typeidea.settings")
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'
# 設(shè)置代理人broker
BROKER_URL = 'redis://127.0.0.1:6379/1'
# celery 的啟動(dòng)工作數(shù)量設(shè)置
CELERY_WORKER_CONCURRENCY = 20
# 任務(wù)預(yù)取功能，就是每個(gè)工作的進(jìn)程／線程在獲取任務(wù)的時(shí)候，會(huì)盡量多拿 n 個(gè)，以保證獲取的通訊成本可以壓縮。
CELERYD_PREFETCH_MULTIPLIER = 20
# 非常重要,有些情況下可以防止死鎖
CELERYD_FORCE_EXECV = True
# celery 的 worker 執(zhí)行多少個(gè)任務(wù)后進(jìn)行重啟操作
CELERY_WORKER_MAX_TASKS_PER_CHILD = 100
# 禁用所有速度限制，如果網(wǎng)絡(luò)資源有限，不建議開(kāi)足馬力。
CELERY_DISABLE_RATE_LIMITS = True
 
CELERY_ENABLE_UTC = False
CELERY_TIMEZONE = settings.TIME_ZONE
DJANGO_CELERY_BEAT_TZ_AWARE = False
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'
 

import time
from celery import Celery
from blog.celery import app
 
# 使用redis做為broker
# app = Celery('blog.tasks2',broker='redis://127.0.0.1:6379/0',backend='redis://127.0.0.1:6379/1')
 
# 創(chuàng)建任務(wù)函數(shù)
@app.task
def my_task(a, b, c):
    print('任務(wù)正在執(zhí)行...')
    print('任務(wù)1函數(shù)休眠10s')
    time.sleep(10)
    return a + b + c
 
@app.task
def my_task2():
    print("任務(wù)2函數(shù)正在執(zhí)行....")
    print('任務(wù)2函數(shù)休眠10s')
    time.sleep(10)

$ celery -A blog worker -l info 

celery -A blog beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler

>>> from django_celery_beat.models import PeriodicTask, IntervalSchedule
>>> schedule, created = IntervalSchedule.objects.get_or_create( 
...       every=10, 
...       period=IntervalSchedule.SECONDS, 
...  )
>>> IntervalSchedule.objects.all()
<QuerySet [<IntervalSchedule: every 10 seconds>]>

>>>PeriodicTask.objects.create(interval=schedule,name='my_task2',task='blog.tasks.my_task2',)
<PeriodicTask: my_task2: every 10 seconds>

>>> PeriodicTask.objects.create(interval=schedule,name='my_task',task='blog.tasks.my_task',args=json.dumps([10,20,30]))
<PeriodicTask: my_task: every 10 seconds>

>>> import pytz
>>> schedule, _ = CrontabSchedule.objects.get_or_create(
... minute='*',
... hour='*',
... day_of_week='*',
... day_of_month='*',
... timezone=pytz.timezone('Asia/Shanghai')
... )

PeriodicTask.objects.create(crontab=schedule,name='my_task2_crontab',task='blog.tasks.my_task2',)

>>> PeriodicTask.objects.all()
<ExtendedQuerySet [<PeriodicTask: celery.backend_cleanup: 0 4 * * * (m/h/dM/MY/d) Asia/Shanghai>, <PeriodicTask: my_task2_crontab: * * * * * (m/h/dM/MY/d) Asia/Shanghai>]>
>>> PeriodicTask.objects.get(name='my_task2_crontab')
<PeriodicTask: my_task2_crontab: * * * * * (m/h/dM/MY/d) Asia/Shanghai>
>>> for task in PeriodicTask.objects.all():
...     print(task.id)
... 
1
13
>>> PeriodicTask.objects.get(id=13)
<PeriodicTask: my_task2_crontab: * * * * * (m/h/dM/MY/d) Asia/Shanghai>
>>> PeriodicTask.objects.get(name='my_task2_crontab')
<PeriodicTask: my_task2_crontab: * * * * * (m/h/dM/MY/d) Asia/Shanghai>

>>> my_task2_crontab = PeriodicTask.objects.get(id=13)
>>> my_task2_crontab.enabled
True
>>> my_task2_crontab.enabled=False
>>> my_task2_crontab.save()

>>> my_task2_crontab.enabled
False
>>> my_task2_crontab.enabled=True
>>> my_task2_crontab.save()

PeriodicTask.objects.get(name='my_task2_crontab').delete()
>>> PeriodicTask.objects.get(name='my_task2_crontab')
Traceback (most recent call last):
  File "<console>", line 1, in <module>
  File "/Users/julius/PycharmProjects/typeidea/.venv/lib/python3.9/site-packages/django/db/models/manager.py", line 85, in manager_method
    return getattr(self.get_queryset(), name)(*args, **kwargs)
  File "/Users/julius/PycharmProjects/typeidea/.venv/lib/python3.9/site-packages/django/db/models/query.py", line 435, in get
    raise self.model.DoesNotExist(
django_celery_beat.models.PeriodicTask.DoesNotExist: PeriodicTask matching query does not exist.