您现在的位置是：网站首页>文章详情文章详情

Django2.1使用celery并接入xadmin

inlike2019-05-31【原创文章】浏览(3110) 评论(1) 喜欢(42)

简介本博客开发中使用了较新的Django2.1和xadmin，其中的坑不可谓不多，从最开始的Django1.9换到2.0再到最后的2.1版本，踩了不少坑，这里记录一下Django2.1使用celery以及在xadmin后台中展示celery任务队列

本博客开发中使用了较新的Django2.1和xadmin，其中的坑不可谓不多，从最开始的Django1.9换到2.0再到最后的2.1版本，踩了不少坑，这里记录一下Django2.1使用celery以及在xadmin后台中展示celery任务队列。

首先Django2.1已经不能使用Django-celery了，而是直接配合最新的celery4.x在加上redis3.2，同时使用django_celery_results来管理celery任务队列；django_celery_results会创建数据库，用于记录celery的运行情况。

Django2.1使用celery

安装依赖包：redis==3.2.0; django-celery-results; celery==4.1.0;eventlet安装完成后在新建的应用目录下创建celery.py文件（如图），并写一下代码：

# -*- coding: utf-8 -*-
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

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

# 注册Celery的APP
app = Celery('newblog')
# 绑定配置文件
app.config_from_object('django.conf:settings', namespace='CELERY')

# 自动发现各个app下的tasks.py文件
app.autodiscover_tasks()

然后在celery.py的同级目录下的__init__.py文件中写下面代码：

from __future__ import absolute_import, unicode_literals
from .celery import app as celery_app

__all__ = ['celery_app']

上面两两步是完成celery创建一个celery应用，接下来时是完成配置和创建分布式应用；然后在setting.py内配置celery如下：

CELERY_RESULT_BACKEND = 'django-db'

CELERY_BROKER_URL = 'redis://auth:*******************'

CELERY_IMPORTS = ['blogapp.task']

force = True  # 很重要，没有的话不会加载settin.py中的部分配置

CELERYD_ACKS_LATE = True  # 设置失败允许重试

CELERYD_TASK_TIME_LIMIT = 12 * 30  # 单个任务的最大运行时间，超时会被杀死

CELERYD_CONCURRENCY = 8  # 设置并发的worker数量

然后我们在项目的应用下创建任务：

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

from celery import task
from django.conf import settings
from django.core.mail import send_mail, send_mass_mail, EmailMultiAlternatives

sender = settings.DEFAULT_FROM_EMAIL


@task
def send_email(msg, receivers: list, title='您在李科博客留言有新的回复消息啦'):
    res = send_mail(title, msg, sender, receivers)
    if res == 1:
        return '发送成功'
    else:
        return '发送失败'

@task
def test(a=4):
    return 8 + a

如果存在定时任务，那么在setting.py文件中继续添加配置：

CELERY_BEAT_SCHEDULE = {
    # 周期性任务
    'task-one': {
        'task': 'blogapp.task.wordcloud',
        'schedule': 60 * 60 * 24,  # 每天执行一次
        # 'args': ()
    },
   # 定时任务
    'task-two': {
        'task': 'app.tasks.sendemail',
        'schedule': crontab(minute=0, hour='*/3,9-11'),
        # 'args': ()
    }
}

然后是启动命令，在manage.py同级目录下运行启动命令：

celery -A newblog worker -l info -P eventlet # 启动worker 
celery -A newblog beat -l info # 启动配置文件中的周期和定时任务

celery任务情况在xadmin中显示

django-celery-results是用于连接celery和Django的第三方库，首先安装，安装之后在setting.py文件中注册应用

注册后执行数据库迁移命令：python manage.py migrate生成相关的数据表及sql语句

接下来就是将celery数据展示在xadmin中，在admim文件中写下面代码：

from django_celery_results.models import TaskResult
class TaskResultInfo:
    date_hierarchy = 'date_done'
    list_display = ('task_id', 'task_name', 'date_done', 'status')
    list_filter = ('status', 'date_done', 'task_name',)
    readonly_fields = ('date_done', 'result', 'hidden', 'meta')
    search_fields = ('task_name', 'task_id', 'status')
    fieldsets = (
        (None, {
            'fields': (
                'task_id',
                'task_name',
                'status',
                'content_type',
                'content_encoding',
            ),
            'classes': ('extrapretty', 'wide')
        }),
        (_('Parameters'), {
            'fields': (
                'task_args',
                'task_kwargs',
            ),
            'classes': ('extrapretty', 'wide')
        }),
        (_('Result'), {
            'fields': (
                'result',
                'date_done',
                'traceback',
                'hidden',
                'meta',
            ),
            'classes': ('extrapretty', 'wide')
        }),
    )
xadmin.site.register(TaskResult, TaskResultInfo)