xadmin引入celery4.0執行異步任務與定時任務

本文轉載自查看原文 2019-10-05 19:20 652 django/ Django（Python3）

一、安裝

pip install celery

pip install django-celery-beat

pip install django-celery-results

pip install redis

安裝這4個庫

pip list

celery==4.3.0
django-celery-beat==1.5.0
django-celery-results==1.1.2
redis==3.2.1

1、請注意不是django-celery、django-redis、celery-with-redis等庫

以前版本的Celery需要一個單獨的庫來與Django一起工作

但是自從3.1以后就不再是這樣了

Django現在支持開箱即用

2、在每個需要用到Celery的文件的第一行加上

from __future__ import absolute_import, unicode_literals

二、配置Celery

1、在項目demo/settings.py文件下加入：

'django_celery_beat',
# django-celery-beat

'django_celery_results',
# django-celery-results

CELERY_ENABLE_UTC = False
# 不使用國際標准時間
CELERY_TIMEZONE = 'Asia/Shanghai'
# 使用亞洲/上海時區
DJANGO_CELERY_BEAT_TZ_AWARE = False
# 解決時區問題
CELERY_BROKER_URL = 'redis://:Abcdef@123456@192.168.1.102:6379/0'
# redis://:password@hostname:port/db_number
CELERY_BROKER_TRANSPORT = 'redis'
# 使用redis作為中間件
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'
# 自定義調度類，使用Django的ORM
CELERY_RESULT_BACKEND = 'django-db'
# 任務結果，使用Django的ORM
CELERY_ACCEPT_CONTENT = ['application/json']
# 設置任務接收的序列化類型
CELERY_TASK_SERIALIZER = 'json'
# 設置任務序列化方式
CELERY_RESULT_SERIALIZER = 'json'
# 設置結果序列化方式

2、在項目demo目錄下新建celery.py文件：

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery, platforms
from django.utils.datetime_safe import datetime

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'demo.settings')
# 設置默認celery命令行的環境變量

app = Celery('demo')
# 實例化celery

app.now = datetime.now
# 解決時區問題

app.config_from_object('django.conf:settings', namespace='CELERY')
# 直接從Django設置中配置Celery

app.autodiscover_tasks()
# 從所有應用中加載任務模塊tasks.py

platforms.C_FORCE_ROOT = True
# 解決celery不能root用戶啟動的問題

3、在項目demo/__init__.py文件加入：

（請注意不是應用的__init__.py）

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

import pymysql

__all__ = ('celery_app',)

pymysql.install_as_MySQLdb()

4、在應用product目錄下面新建tasks.py文件：

# Create your tasks here
from __future__ import absolute_import, unicode_literals

from demo.celery import app


@app.task
def debug_1():
 print("調試_1已運行！")


@app.task
def debug_2():
 print("調試_2已運行！")

5、在應用product目錄下面product/admin.py文件加入：

from django_celery_beat.models import IntervalSchedule, CrontabSchedule, ClockedSchedule, SolarSchedule, PeriodicTask
from django_celery_results.models import TaskResult

class IntervalScheduleAdmin(object):
 list_display = [
 'id', 'every', 'period',
 ]
 ordering = ['id']
 search_fields = ['every']
 list_per_page = 10


class CrontabScheduleAdmin(object):
 list_display = [
 'id', 'minute', 'hour',
 'day_of_week', 'day_of_month', 'month_of_year', 'timezone'
 ]
 ordering = ['id']
 search_fields = ['minute']
 list_per_page = 10


class SolarScheduleAdmin(object):
 list_display = [
 'id', 'event', 'latitude', 'longitude'
 ]
 ordering = ['id']
 search_fields = ['event']
 list_per_page = 10


class ClockedScheduleAdmin(object):
 list_display = [
 'id', 'clocked_time', 'enabled'
 ]
 ordering = ['id']
 search_fields = ['clocked_time']
 list_per_page = 10


class PeriodicTaskAdmin(object):
 list_display = [
 'id', 'name', 'task', 'args', 'kwargs', 'queue',
 'exchange', 'routing_key', 'expires', 'enabled',
 'last_run_at', 'total_run_count', 'date_changed', 'description',
 'interval', 'crontab', 'solar', 'clocked', 'one_off',
 'start_time', 'priority', 'headers'
 ]
 ordering = ['id']
 search_fields = ['name']
 list_per_page = 10


class TaskResultAdmin(object):
 list_display = [
 'id', 'task_id', 'status', 'content_type', 'content_encoding',
 'result', 'date_done', 'traceback', 'hidden', 'meta',
 'task_args', 'task_kwargs', 'task_name'
 ]
 ordering = ['id']
 search_fields = ['task_id']
 list_per_page = 10

xadmin.site.register(IntervalSchedule, IntervalScheduleAdmin)
# 間隔時間表
xadmin.site.register(CrontabSchedule, CrontabScheduleAdmin)
# 定時時間表
xadmin.site.register(SolarSchedule, SolarScheduleAdmin)
# 太陽時間表
xadmin.site.register(ClockedSchedule, ClockedScheduleAdmin)
# 計時時間表
xadmin.site.register(PeriodicTask, PeriodicTaskAdmin)
# 配置任務

xadmin.site.register(TaskResult, TaskResultAdmin)
# 任務結果