django定时任务库 django_apscheduler

django_apscheduler 在做定时任务的时候这个库挺有用的，记得在这个库还不存在的时候，自己也写了一个和这个库功能差不多的django模块。也能用，不过不是很稳定，每次重启wsgi的时候，都要手动启动任务。无意中发现了django_apscheduler 这块第三方库，用了下挺好用的。

• 安装

pip install django-apscheduler 

• 将： django_apscheduler 加到 项目setting.py INSTALLED_APPS 中

INSTALLED_APPS = ( 
      ... 
      django_apscheduler, 
)

• 设置后需要同步数据库，执行：

python manage.py migrate

会创建两张表：
django_apscheduler_djangojob
django_apscheduler_djangojobexecution

下面就是写逻辑代码了，需要注意的是，定时器需要写在urls里面，任何模块的urls里面都可以。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time    : 2020/7/19 11:41
# @Author  : 熊利宏
# @project : 定时任务
# @Email   : [e-mail]
# @File    : timed_task.py
# @IDE     : PyCharm
# @REMARKS : 定时任务

# 定时任务库
from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore, register_events, register_job

# 综合模块
from WanBaoScience import utility_function

# xToolkit
from xToolkit import xdatetime

# django 配置文件
from WanBaoScience import settings


# 每天 12,18,21 点执行
def job_one():
    utility_function.enterprise_mailbox("G", "这是每天 12时，18时，21时执行的定时任务，当前时间{}-{}".format(xdatetime.get().format(), settings.ENVIRONMENT))


# 每天 0点5分 执行
def test_two():
    utility_function.enterprise_mailbox("G", "这是每天 0点5分 执行的定时任务，当前时间{}-{}".format(xdatetime.get().format(), settings.ENVIRONMENT))


# 每30秒 执行
def test_three():
    pass
    # utility_function.enterprise_mailbox("G", "这是每30秒执行的定时任务，当前时间{}-{}".format(xdatetime.get().format(), settings.ENVIRONMENT))


# 每小时执行的任务
def job_four():
    utility_function.enterprise_mailbox("G", "这是每小时执行的定时任务，当前时间{}-{}".format(xdatetime.get().format(), settings.ENVIRONMENT))


# 每10分钟 执行
def test_five():
    utility_function.enterprise_mailbox("G", "这是每10分钟执行的定时任务，当前时间{}-{}".format(xdatetime.get().format(), settings.ENVIRONMENT))


# 每1分钟 执行
def test_six():
    pass
    # utility_function.enterprise_mailbox("G", "这是每1分钟执行的定时任务，当前时间{}-{}".format(xdatetime.get().format(), settings.ENVIRONMENT))


if settings.ENVIRONMENT == "official":
    # 锁库
    import atexit
    import fcntl

    f = open("scheduler.lock", "wb")
    try:
        fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
        # 实例化调度器
        scheduler = BackgroundScheduler()
        scheduler.add_jobstore(DjangoJobStore(), "default")
        scheduler.add_job(job_one, "cron", hour='12,18,21', id='job_one')  # 每天 12,18,21 点执行
        scheduler.add_job(test_two, "cron", hour=0, minute=5, second=0, id='test_two')  # 每天 0点5分 执行
        scheduler.add_job(test_three, "cron", second="0/30", id='test_three')  # 每30秒 执行
        scheduler.add_job(job_four, "cron", hour='0/1', minute=0, second=0, id='job_four')  # 每小时执行的任务
        scheduler.add_job(test_five, "cron", minute='0/10', id='test_five')  # 每10分钟 执行
        scheduler.add_job(test_six, "cron", minute='0/1', id='test_six')  # 每1分钟 执行
        register_events(scheduler)
        scheduler.start()
    except BaseException:
        pass


    def unlock():
        fcntl.flock(f, fcntl.LOCK_UN)
        f.close()


    atexit.register(unlock)

• 一些自己踩过的坑：
  1.django_apscheduler + uwsgi 同一个定时任务重复执行 如何解决？

解决这个问题，网上面有三种方法：
1.设置uwsgi 的工作线程为一个
2.用 锁
3.端口方法，比较另类，定时简单实用。我就是用的这个方法，见上面代码。

• 参考资料：

interval/cron触发器详解 常用

interval 触发器
参数 说明
weeks (int) 间隔几周
days (int) 间隔几天
hours (int) 间隔几小时
minutes (int) 间隔几分钟
seconds (int) 间隔多少秒
start_date (datetime 或 str) 开始日期
end_date (datetime 或 str) 结束日期
timezone (datetime.tzinfo 或str) 时区

每隔两分钟执行一次 job_func 方法
scheduler .add_job(job_func, ‘interval’, minutes=2)
在 2017-12-13 14:00:01 ~ 2017-12-13 14:00:10 之间, 每隔两分钟执行一次 job_func 方法
scheduler .add_job(job_func, ‘interval’, minutes=2, start_date=‘2017-12-13 14:00:01’ , end_date=‘2017-12-13 14:00:10’)

cron 触发器:
参数 说明
year (int 或 str) 年，4位数字
month (int 或 str) 月 (范围1-12)
day (int 或 str) 日 (范围1-31
week (int 或 str) 周 (范围1-53)
day_of_week (int 或 str) 周内第几天或者星期几 (范围0-6 或者 mon,tue,wed,thu,fri,sat,sun)
hour (int 或 str) 时 (范围0-23)
minute (int 或 str) 分 (范围0-59)
second (int 或 str) 秒 (范围0-59)
start_date (datetime 或 str) 最早开始日期(包含)
end_date (datetime 或 str) 最晚结束时间(包含)
timezone (datetime.tzinfo 或str) 指定时区

在每年 1-3、7-9 月份中的每个星期一、二中的 00:00, 01:00, 02:00 和 03:00 执行 job_func 任务
scheduler .add_job(job_func, ‘cron’, month=‘1-3,7-9’,day=‘0, tue’, hour=‘0-3’)