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()
打印格式化的任务列表
服务器托管,北京服务器托管,服务器租用 http://www.fwqtg.net