Django项目中使用Celery

pip install celery

# -*- coding: utf-8 -*-
from __future__ import absolute_import

import os

from celery import Celery, platforms

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

# from django.conf import settings

platforms.C_FORCE_ROOT = True   # 允许用root用户启动celery
# 实例化Celery
app = Celery('my_site')

# Using a string here means the worker will not have to
# pickle the object when using Windows.
# 使用django的settings文件配置celery
app.config_from_object('django.conf:settings', namespace='CELERY')
# Celery加载所有注册的应用
app.autodiscover_tasks()

# -*- coding: utf-8 -*-
from __future__ import absolute_import

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
# 引入celery实例对象
from .celery import app as celery_app
__all__ = ['celery_app']

# -*- coding: utf-8 -*-

from __future__ import absolute_import
from celery.schedules import crontab

CELERY_WORKER_MAX_TASKS_PER_CHILD = 100000  # 每个worker执行10w个任务就会被销毁，可防止内存泄露
# celery中间人 redis://redis服务所在的ip地址:端口/数据库号
CELERY_BROKER_URL = "redis://127.0.0.1:6379/3"
# celery结果返回，可用于跟踪结果
CELERY_RESULT_BACKEND = "redis://127.0.0.1:6379/4"
CELERY_RESULT_EXPIRES = 3600 * 24  # 任务多久被清除
CELERY_TASK_IGNORE_RESULT = False  # 一般不关注结果，请开启该设置，如果要存结果，请配置CELERY_RESULT_BACKEND
# celery时区设置，使用settings中TIME_ZONE同样的时区
CELERY_TIMEZONE = 'Asia/Shanghai'

CELERY_TASK_DEFAULT_EXCHANGE = 'mysite_exchange'
CELERY_TASK_DEFAULT_EXCHANGE_TYPE = 'direct'
CELERY_TASK_DEFAULT_QUEUE = 'mysite_queue'  # 默认是celery,一般修改
CELERY_TASK_DEFAULT_ROUTING_KEY = 'default'

# crotab任务，这里是下面即将出现的定时任务
CELERY_BEAT_SCHEDULE = {
    'submit_urls_to_baidu': {
        'task': 'article.tasks.submit_urls_to_baidu',
        'schedule': crontab(minute=0, hour="*/1"),  # 每小时执行
    },
}

# -*- coding: utf-8 -*-

import requests
from article.models import Article
from article.constants import BlogStatus, DOMAIN
from celery import shared_task
from utils.libs.logger.syslogger import SysLogger


@shared_task
def submit_urls_to_baidu():
    articles = Article.objects.filter(status=BlogStatus.PUBLISHED).order_by('-id')
    urls = [DOMAIN + article.get_absolute_url() for article in articles]
    api = 'http://data.zz.baidu.com/urls?site=yangsihan.com&token=xxxxxxxxxx'
    response = requests.post(api, data='\n'.join(urls))
    SysLogger.info(response.content.decode())
    print (response.content.decode())

# coding:utf-8
from celery.decorators import task 
import time

@task
def sendmail(email):
    print('start send email to %s' % email)
    time.sleep(5) #休息5秒
    print('success')
    return True

# coding:utf-8
from django.shortcuts import render
from django.http import HttpResponse

from .models import Article
from .tasks import sendmail #引用tasks.py文件的中sendmail方法
import json

def home(request):
    # 耗时任务，发送邮件（用delay执行方法）
    sendmail.delay('test@test.com')

    # 其他行为
    data = list(Article.objects.values('name'))
    return HttpResponse(json.dumps(data), content_type = 'application/json')

Celery -A my_site worker -l info  # my_site是项目名

Celery -A my_site beat -l info