# 定时任务例子

**Repository Path**: webshare520/examples_of_timing_tasks

## Basic Information

- **Project Name**: 定时任务例子
- **Description**: django2+xadmin+celery+django_celery+beat实现的定时任务管理
- **Primary Language**: Python
- **License**: AFL-3.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 2
- **Created**: 2023-07-08
- **Last Updated**: 2023-07-08

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

---
title: django2+xadmin+celery实现定时周期任务 
tags: django,xadmin,python,celery,django_celery_beat
grammar_cjkRuby: true
---


# 基本环境

 - 开发环境：windows 10
 - python版本：3.6.5
 - django版本：2.1.4
 
 >这里我已经创建了一个实例环境了，基本的django2+xadmin并且能正常运行了。以下是基于该环境写的。

当前示例文件结构如下：
![enter description here](https://gitee.com/q315593255/xiaoshujiang/raw/master/2019524/1558665591367.png)

# 添加定时任务
目前celery的版本已经是4.x了，而djcelery仅支持3.x的版本，因此需要使用django_celery_beat来实现定时周期的任务。

 1. 安装调度用的模块
 ```
 pip install celery==4.3.0
 pip install django_celery_beat==1.5.0
 ```
 2. 修改项目settings.py设置
```
INSTALLED_APPS = [
   #...
    'django_celery_beat',
]

BROKER_URL = 'sqla+sqlite:///db.sqlite3'
CELERY_BROKER_URL = 'sqla+sqlite:///db.sqlite3'
CELERY_TIMEZONE = 'Asia/Shanghai'

CELERY_ACCEPT_CONTENT = ['pickle', 'json','application/json']
CELERYD_CONCURRENCY = 8                                          # 并发worker数
CELERYD_FORCE_EXECV = True                                       # 非常重要,有些情况下可以防止死锁
CELERYD_MAX_TASKS_PER_CHILD = 100    # 每个worker最多执行100个任务就会被销毁，可防止内存泄露
CELERY_DISABLE_RATE_LIMITS = True    # 任务发出后，经过一段时间还未收到acknowledge , 就将任务重新交给其他worker执行

CELERY_BEAT_SCHEDULER='django_celery_beat.schedulers.DatabaseScheduler'
CELERYBEAT_SCHEDULER='django_celery_beat.schedulers.DatabaseScheduler'
```

 3. 在和同settings.py目录下增加celery.py文件
```
#celery.py
from __future__ import absolute_import
 
import os
from celery import Celery, platforms
from django.conf import settings
 
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'examples_of_timing_tasks.settings') #examples_of_timing_tasks是项目名称
 
app = Celery('examples_of_timing_tasks')
platforms.C_FORCE_ROOT = True
 
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
app.conf.update(
    BROKER_URL=settings.BROKER_URL,

)
 
@app.task(bind=True)
def test(self,info):
    print(info)
    print('222222222222222222222222222')
```

 4.  \_\_init\_\_.py文件增加以下内容：
```
from __future__ import absolute_import, unicode_literals
from .celery import app as celery_app
__all__ = ['celery_app']
```
5. 在已注册的app下创建adminx.py文件，然后将定时任务相关的models注册到后台管理去
```
#adminx.py
from __future__ import absolute_import, unicode_literals
import xadmin
from django_celery_beat.models import (
  PeriodicTask, 
  IntervalSchedule, 
  CrontabSchedule,
)
#celery
xadmin.site.register(IntervalSchedule) # 存储循环任务设置的时间
xadmin.site.register(CrontabSchedule) # 存储定时任务设置的时间
xadmin.site.register(PeriodicTask) # 存储任务

```
到这里之后需要重新生成一下相关表结构然后重新运行一下
```
python manage.py migrate
python manage.py runserver
```
打开网站就可以看到
![enter description here](https://gitee.com/q315593255/xiaoshujiang/raw/master/2019524/1558667633876.png)

这里框架基本上是完成了，然后我们启动一下下面的这两个任务
```
celery -A examples_of_timing_tasks beat -l info
celery -A examples_of_timing_tasks worker -l info -P eventlet
```
#### 创建测试任务

 1. 先创建一个定时器![enter description here](https://gitee.com/q315593255/xiaoshujiang/raw/master/2019524/1558667988790.png)
 2. 创建任务![enter description here](https://gitee.com/q315593255/xiaoshujiang/raw/master/2019524/1558668084366.png)
 3. 可以在任务监控平台看到任务已经运行起来了![enter description here](https://gitee.com/q315593255/xiaoshujiang/raw/master/2019524/1558668174245.png)