自学Python第二十二天- Django框架(六) django的实用插件：cron、APScheduler
django-crontab 和 django-cron
有时候需要django在后台不断的执行一个任务，简单的可以通过中间件来实现，但是中间件是根据请求触发的。如果需要定时执行任务，则需要使用到一些插件。
django-crontab 和 django-cron 是常用的用于处理定时任务的插件库，两者区别在于：
- django-cron 完全运行在 django 服务器内部的库，它通过定期检查当前时间与定义的任务计划是否匹配来执行任务。它不依赖于系统级的cron守护进程，而是基于 django 自身的请求周期触发任务检查
- django-crontab 是利用操作系统的 cron 守护进程来调度任务。用户在 django 项目中定义好任务后， django-crontab 会将这些任务写入到系统的 crontab 文件中。当 cron 守护进程执行任务时，实际上会调用 django 管理命令或 url 来执行特定的操作。
django-crontab
使用 django-crontab 需要以下步骤：
- 安装 django-crontab 模块
- 在 settings.py 中注册应用
- 创建定时执行的任务(可以是自定义的函数或自定义的命令)
- 在 settings.py 中配置执行任务
- 执行任务
需要注意的是 django-crontab 只能运行在 linux 环境中，且需要使用到 root 权限。另外需要启动 cron 服务
# 查看 cron 服务状态
service cron status
# 开启 cron 服务
service cron start
# 查看定时任务
crontab -l
# 添加定时任务
crontab -e
安装 django-crontab
pip install django-crontab
注册应用
在 settings.py 中注册 django-crontab 应用
INSTALLED_APPS = [
...
'django_crontab',
]
创建任务
创建定时执行的任务
创建定时执行函数
在任意 app 下均可创建定时执行函数
# app1/task.py
def scheduleTask(): # 定时执行函数
from time import strftime, localtime
print(strftime("%Y-%m-%d %H:%M:%S", localtime()), end='')
print("执行了scheduleTask函数")
def scheduleTaskWithPara(in_str): # 带参数的定时执行函数
import datetime
now_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
print(f"在 {now_time} 执行了定时函数，输入信息{in_str}")
创建定时执行命令
在任意 app 下创建 management 模块(含有 __init__.py 文件)，并在此模块下创建 commands 模块。在 commands 模块下创建具体的执行脚本文件。例如：
# app1/management/commands/mycommand1.py
from django.core.management.base import BaseCommand, CommandError
from time import strftime, localtime
class Command(BaseCommand):
help = '这是第一个command测试指令'
# 为handle中添加参数解析，
def add_arguments(self, parser):
parser.add_argument(
'-p', # 设置参数的时候 前边携带
'--param',
action='store',
dest='param', # 自定义传入的参数键名ss
default='close', # 默认的键值
help='name of author.',
)
def handle(self, *args, **options):
# print("mycommand1----开始")
'''
添加你需要功能，（访问数据库，判断有效性等等）...
'''
# 例如
try:
if options['param']:
print(strftime("%Y-%m-%d %H:%M:%S", localtime()), end='')
print(": mycommand1传入的参数为", options['param'])
except Exception as e:
print("12",e)
print(CommandError("1111111111111111111111111"))
# print("mycommand1----结束")
# app1/management/commands/mycommand2.py
from django.core.management.base import BaseCommand
from time import strftime, localtime
class Command(BaseCommand):
help = '这是第二个command测试指令'
# 为handle中添加参数解析，
def add_arguments(self, parser):
parser.add_argument(
'-p', # 设置参数的时候 前边携带
'--param',
action='store',
dest='param', # 自定义传入的参数键名ss
default='close', # 默认的键值
help='name of author.',
)
def handle(self, *args, **options):
# print("mycommand2----开始")
'''
添加你需要功能，（访问数据库，判断有效性等等）...
'''
# 例如
if options['param']:
print(strftime("%Y-%m-%d %H:%M:%S", localtime()), end='')
print("mycommand2传入的参数为", options['param'])
# print("mycommand2----结束")
配置任务
将创建好的定时执行任务配置到 settings.py 中去，以方便执行
配置定时执行函数
在 CRONJOBS 列表中添加一个元组，第一个元素是执行间隔时间，第二个元素是执行任务函数。如没有参数第三个元素是输出处理，如有参数第三、四元素分别为传参，第五参数是输出处理。例如：
CRONJOBS = [
# 每1分钟执行scheduleTask函数，并将执行中的返回的内容全部打印到crontab.log文件中
('*/1 * * * *', 'app1.tasks.scheduleTask', '>> /tmp/crontab.log'),
('*/1 * * * *', 'app1.tasks.scheduleTaskWithPara', ['James'], {}, '>> /tmp/crontab.log'),
]
配置定时执行命令
类似于定时执行函数，将定时执行命令添加到 CRONJOBS 列表中去。区别在于元组的元素定义不同
CRONJOBS = [
# 每1分钟执行django的自定义命令，并将执行中的返回的内容全部打印到crontab.log文件中
('*/1 * * * *', 'django.core.management.call_command', ['mycommand1'], {"param": "mycommand1_test"}, '>>/home/wangzhipeng/myproject/crontab.log'),
('*/1 * * * *', 'django.core.management.call_command', ['mycommand2'], {"param": "mycommand2_test"}, '>>/home/wangzhipeng/myproject/crontab.log')
]
- 元素1：定时 例如*/1 * * * * 表示每隔1分钟执行
- 元素2：方法的python模块路径，如果执行django-admin命令，则写django.core.management.call_command
- 元素3：方法的位置参数列表（默认值：[]），如果执行django-admin命令，则填写所需执行的命令
- 元素4：方法的关键字参数的dict（默认值：{}）
- 元素5：执行log存放位置（即重定向到文件，默认：‘’）
配置完成后可以进行测试：
python manage.py mycommand1 -p 123
# 2019-05-10 15:10:59: mycommand1传入的参数为 123
关于 linux 的 crontab 时间语法
crontab 的时间语法由五部分组成，通常为五个 *，每一位表示的意思分别为：分钟、小时、日、月、星期。
可使用的字符和其意义为：
| 字符 | 含义 |
|---|---|
* | 代表所有取值范围的数字 |
/ | 代表“每”的意思，例如 */5 表示5个单位 |
- | 代表从某个数字到某个数字 |
, | 分隔离散的数字 |
例如：
| 示例 | 含义 |
|---|---|
0 */2 * * * | 每2个小时 |
0 23-7,9 * * * | 23点到7点，或9点 |
0 11 4 * 1-3 | 每月4号或周一到三的11点 |
0 4 1 1 * | 1月1日4点 |
0 6 * * * | 每天6点 |
0 */2 * * * | 每2小时 |
执行任务
添加并开启定时任务：
python manage.py crontab add
查看执行中的定时任务
python manage.py crontab show
删除(停止)定时任务
python manage.py crontab remove
django-cron
django-cron 因为完全基于 django，使得使用简单。但是也是这个原因，要想使定时任务生效，django服务必须一直处于运行状态。
另外 django-cron django-cron会依赖于HTTP请求来触发其内部的任务检查逻辑（通过中间件实现）。这意味着只有当有用户访问网站或通过其他方式触发HTTP请求时，cron任务才可能被执行。
需注意的是，dnango-cron 也只能在 linux 系统上运行。
(django-cron英文文档)[https://django-cron.readthedocs.io/en/latest/]
使用 django-cron 需要以下步骤：
- 安装 django-cron 模块
- 在 settings.py 中注册应用
- 进行迁移，创建 django_cron 应用所需资源
- 创建定时执行的任务类
- 在 settings.py 中配置执行任务
安装 django-cron
pip install django-cron
注册应用
在 settings.py 中注册应用：
INSTALLED_APPS = [
# ...
"django_cron",
]
创建 django_cron 资源
python manage.py migrate django_cron
创建任务类
和 django-crontab 不同的是， django-cron 的定时任务是一个集成自 django_cron.CronJobBase 的类
# app1/cron.py
from django_cron import CronJobBase, Schedule
from django.core.management import call_command
class MyCronJob(CronJobBase):
RUN_EVERY_MINS = 30 # 每30分钟运行一次
schedule = Schedule(run_every_mins=RUN_EVERY_MINS) # 添加任务执行周期
code = 'yourapp.my_cron_job' # 唯一标识符
def do(self):
# 执行的命令或函数
call_command('your_management_command') # 如果是管理命令
# 或执行任意Python代码
my_function()
如果需要在指定时间执行任务，而不是按周期执行，则使用schedule = Schedule(run_at_times=['11:30', '14:00', '23:15'])，注意必须使用24小时制。按照周期执行和执行时间执行可以一起使用。schedule = Schedule(run_every_mins=RUN_EVERY_MINS, run_at_times=RUN_AT_TIMES)。
配置任务
需要在 settings.py 中的 CRON_CLASSES 列表中注册此任务
CRON_CLASSES = [
"app1.cron.MyCronJob",
]
执行任务
在 django-cron 中，定时任务是在 django 应用运行时触发执行的。如果需要立即运行所有的 cron 任务，例如部署新代码后或进行测试时，可以手动执行任务。
python manage.py runcrons
也可以单独指定手动执行具体的哪个任务类
python manage.py runcrons "app1.cron.MyCronJob" "app1.cron.OtherCronJob"
如果需要静默执行，即不输出相关信息，可以添加参数 --silent。
如果需要强制执行，可以添加参数 --force。
如果需要查看有哪些任务可以执行，而并不真的运行这些任务，可以添加参数 --dry-run。
django-apscheduler
cron 是基于 unix_like 系统的，windows 下使用 APScheduler。APscheduler全称Advanced Python Scheduler，它是一个轻量级的 Python 定时任务调度框架。APScheduler 支持三种调度任务：固定时间间隔，固定时间点（日期），Linux 下的 Crontab 命令。同时，它还支持异步执行、后台执行调度任务。
django-apscheduler 是 APScheduler 的django定制封装插件版，专为 django 设计。它的使用流程为：
- 安装
- 注册应用
- 配置
- 执行迁移
- 创建任务
- 初始化调度器，并添加任务
- 启动任务
apscheduler英文文档
参考文档：定时任务框架APScheduler学习
安装
pip install django-apscheduler
注册应用
在 settings.py 中注册 app
INSTALLED_APPS = [
# ...
'django_apscheduler',
]
执行迁移
python manage.py migrate
APScheduler 会在数据库中创建2张表
- django_apscheduler_djangojob : 用于存储定时任务
- django_apscheduler_djangojobexecution : 用于存储每次的执行记录、时长和执行结果等。
创建任务
apscheduler 的任务就是普通的函数，可以将需要执行的内容写在函数内部，然后使用调度器添加任务即可
def scheduler_task():
pass # 这里是任务需要执行的内容
创建调度器，添加任务
django_apscheduler 是基于 APScheduler 使用的，安装了 django_apscheduler 会自动安装 APScheduler。调度器需要使用 apscheduler.schedulers.backgroud.BackgroundScheduler 或 apscheduler.schedulers.blocking.BlockingScheduler 两个类中的一个来实例化调度器。两者区别在于 BackgroundScheduler 会在后台执行，而 BlockingScheduler 会在主线程执行，进而阻塞主线程。因此常用 BackgroundScheduler 来创建调度器。
from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore, register_job, register_events
def init_scheduler():
scheduler = BackgroundScheduler() # 创建调度器
try:
scheduler.add_jobstore(DjangoJobStore(), 'default') # 添加调度器作业存储
# 添加定时任务，每天0点，30分执行一次，任务id为test。
scheduler.add_job(scheduler_test, 'cron', hour=0, minute=30, id='test', replace_existing=True)
# 添加定时任务，间隔3秒执行一次
scheduler.add_job(scheduler_jog, 'interval', seconds=3, id='3_second_job')
# 用装饰器的方式添加任务
@regist_job(scheduler, "interval", seconds=10, replace_existing=True, id="task_per_10s")
def task_per_10_seconds():
print('间隔10秒执行任务')
# 注册处理事件
register_events(scheduler)
# 调度器开始工作，计时器开始计时
scheduler.start()
except Exception as e:
print(e)
# 报错则调度器终止运行，否则会跳过错误任务继续执行
scheduler.shutdown()
scheduler.add_job(func, trigger=None, args=None, kwargs=None, id=None, name=None, misfire_grace_time=undefined, coalesce=undefined, max_instances=undefined, next_run_time=undefined, jobstore='default', executor='default', replace_existing=False, **trigger_args) 方法可以将任务添加至调度器，其参数有：
- func 必须参数，需要执行的任务函数(或可调用对象)
- trigger 必须参数，定时器类型，可以是触发器实例或触发器字符串
- args 和 kwargs：传递给任务函数的参数列表和关键字参数字典。主要用于trigger使用字符串时，传递具体的定时时间。
- id 和 name：分别用于唯一标识和命名一个任务。如果未提供，将自动生成。
- misfire_grace_time：任务错过执行的时间窗口，在这个时间段内，即使错过了计划执行时间点，也会被执行。
- coalesce：布尔值，表示当任务被错过多次后是否合并执行（只执行一次而不是多次）。
- max_instances：同一时间内允许的最大并发执行实例数。
- next_run_time：手动指定下一次运行时间（通常由调度器自动计算）。
- jobstore：指定存储此任务的作业存储。通常配置为 ‘default’ 或 DjangoJobStore 的实例，且在
add_job之前使用add_jobstore方法配置。 - executor：执行器实例，负责实际执行作业。默认使用已配置好的执行器。
- replace_existing：是否替换同名或同ID的任务，解决第二次调度任务时因为第一次任务未完成而报错的问题
APScheduler 支持的定时器主要有：
| 定时器 | 说明 |
|---|---|
cron | 基于 unix_like 系统的 cron 定时任务 |
interval | 重复定时器 |
date | 固定时间定时器 |
scheduler.add_job() 添加定时器时，可以在第二个参数以字符串形式指定定时器类型，然后使用关键字传参的方式传入具体时间，也可以使用apscheduler.triggers.cron.CronTrigger、apscheduler.triggers.interval.IntervalTrigger 或 apscheduler.triggers.date.DateTrigger 创建定时器对象实例传入第二个参数(或用关键字传参传给trigger)。
cron 定时器：
interval 定时器：
date 定时器
启动任务
在 django 中执行调度器即可开始定时任务，网上很多推荐添加到 urls.py 中执行。如果定时器中需要涉及 django 资源，例如需要使用 app 的 models 来操作数据库，则需要相应资源加载完成。这里推荐写到 app 的 apps.py 中，在设置类的 ready() 方法中，可以在 app 准备完成时执行。
# app1/apps.py
from django.apps import AppConfig
class App1Config(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'app1'
def ready(self):
super().ready()
run_once = os.environ.get('CMDLINE_DJANGO_RUNNER_RUN_ONCE')
if run_once is not None:
return
os.environ['CMDLINE_DJANGO_RUNNER_RUN_ONCE'] = 'True'
# 在这里执行 APScheduler 定时任务
需要注意的是，启动任务前必须完成迁移工作，否则 django_apscheduler 会因数据库中没有创建2张表而报错。
另外 django 默认 runserver 时会创建一个守护进程来监控代码有没有改变以计时重启服务，这在调试时是非常方便的，但是定时任务也会重复执行。所以启用执行任务时需使用 python manage.py runserver --noreload ，或在环境变量(不能在django服务中，因为django进程间数据隔离)添加一个状态量，作为锁使用，防止重复执行。
其他
scheduler 对象还有一些常用方法：
scheduler.remove_job(job_id)删除任务scheduler.pause_job(job_id)暂停任务scheduler.resume_job(job_id)继续执行暂停任务scheduler.modify_job(job_id)修改定时任务(可以修改除了id外的属性，例如定时器等参数)scheduler.get_jobs()获取所有定时任务scheduler.get_job(job_id)获取特定任务scheduler.print_jobs()打印格式化的任务列表