Django项目中使用Celery执行异步和周期性任务

import os
from celery import Celery

# 设置环境变量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

# 实例化
app = Celery('myproject')

# namespace='CELERY'作用是允许你在Django配置文件中对Celery进行配置
# 但所有Celery配置项必须以CELERY开头，防止冲突
app.config_from_object('django.conf:settings', namespace='CELERY')

# 自动从Django的已注册app中发现任务
app.autodiscover_tasks()

# 一个测试任务
@app.task(bind=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

from .celery import app as celery_app
__all__ = ('celery_app',)

# 最重要的配置，设置消息broker,格式为：db://user:password@host:port/dbname
# 如果redis安装在本机，使用localhost
# 如果docker部署的redis，使用redis://redis:6379
CELERY_BROKER_URL = "redis://127.0.0.1:6379/0"

# celery时区设置，建议与Django settings中TIME_ZONE同样时区，防止时差
# Django设置时区需同时设置USE_TZ=True和TIME_ZONE = 'Asia/Shanghai'
CELERY_TIMEZONE = TIME_ZONE

# myproject/tasks.py
# 专属于myproject项目的任务
app = Celery('myproject'）
@ app.task
def test()：
    pass

# app/tasks.py, 可以复用的task
from celery import shared_task
import time

@shared_task
def add(x, y):
    time.sleep(2)
    return x + y

# 方法一：delay方法
task_name.delay(args1, args2, kwargs=value_1, kwargs2=value_2)

# 方法二： apply_async方法，与delay类似，但支持更多参数
task.apply_async(args=[arg1, arg2], kwargs={key:value, key:value})

# app/views.py
from .tasks import add

def test_celery(request):
    add.delay(3, 5)
    return HttpResponse("Celery works")

# app/urls.py
urlpatterns = [
    re_path(r'^test/$', views.test_celery, name="test_celery")
]

# app/views.py
from .tasks import add

def test_celery(request):
    result = add.apply_async(args=[3, 5])
    return HttpResponse(result.task_id + ' : ' + result.status)

# 支持数据库django-db和缓存django-cache存储任务状态及结果
# 建议选django-db
CELERY_RESULT_BACKEND = "django-db"
# celery内容等消息的格式设置，默认json
CELERY_ACCEPT_CONTENT = ['application/json', ]
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

from celery.result import AsyncResult
# 调用异步任务
async_task = add.apply_async(args=[3, 5])
# 获取任务状态和结果
AsyncResult(async_task.task_id).status
AsyncResult(async_task.task_id).result

from datetime import timedelta
CELERY_BEAT_SCHEDULE = {
    "add-every-30s": {
        "task": "app.tasks.add",
        'schedule': 30.0, # 每30秒执行1次
        'args': (3, 8) # 传递参数-
    },
    "add-every-day": {
        "task": "app.tasks.add",
        'schedule': timedelta(hours=1), # 每小时执行1次
        'args': (3, 8) # 传递参数-
    },
}

CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

CELERY_BEAT_SCHEDULE = {
    # 每周一早上7点半执行
    'add-every-monday-morning': {
        'task': 'app.tasks.add',
        'schedule': crontab(hour=7, minute=30, day_of_week=1),
        'args': (7, 8),
    },
}

# 开启任务调度器
Celery -A myproject beat

# Linux下开启Celery worker
Celery -A myproject worker -l info

# windows下开启Celery worker
Celery -A myproject worker -l info -P eventlet

# windows下如果报Pid错误
Celery -A myproject worker -l info --pool=solo

pip install flower

@shared_task(bind=True, max_retries=3)
def send_batch_notifications(self):
   try:
       something_raising()
       raise Exception('Can\'t send email.')
   except Exception as exc:
       self.retry(exc=exc, countdown=5)
   send_mail(
       subject='Batch email notifications',
       message='Test email',
       from_email='no-reply@example.com',
       recipient_list=['john@example.com']
   )

CELERY_TASK_DEFAULT_QUEUE = 'default'
CELERY_TASK_DEFAULT_ROUTING_KEY = 'default'
CELERY_QUEUES = (
   Queue('default', Exchange('default'), routing_key='default'),
   Queue('heavy_tasks', Exchange('heavy_tasks'), routing_key='heavy_tasks'),
)
CELERY_TASK_ROUTES = {
   'myapp.tasks.heave_tasks': 'heavy_tasks'
}

# 话例子
@app.task
def update_page_info(url):
    page = fetch_page.delay(url).get()
    info = parse_page.delay(url, page).get()
    store_page_info.delay(url, info)
 
@app.task
def fetch_page(url):
    return myhttplib.get(url)
 
@app.task
def parse_page(url, page):
    return myparser.parse_document(page)
 
@app.task
def store_page_info(url, info):
    return PageInfo.objects.create(url, info)

class Article(models.Model):
    title = models.CharField()
    body = models.TextField()
 
@app.task
def expand_abbreviations(article):
    article.body.replace('Old text', 'New text')
    article.save()

@app.task
def expand_abbreviations(article_id):
    article = Article.objects.get(id=article_id)
    article.body.replace('MyCorp', 'My Corporation')
    article.save()

from django.db import transaction
 
@transaction.commit_on_success
def create_article(request):
    article = Article.objects.create()
    expand_abbreviations.delay(article.pk)

from django.db.transaction import on_commit
 
def create_article(request):
    article = Article.objects.create()
    on_commit(lambda: expand_abbreviations.delay(article.pk))