Python开发—14、定时任务

• 1、需求简介
• 2、类库实现
• • 2.1、while...sleep
  • 2.2、threading.Timer
  • 2.3、schedule模块
  • 2.4、APScheduler框架
  • • 2.4.3、触发器(trigger)

在实际的应用开发中，很多场景下需要在固定的时间，按照固定的时间间隔周期性的运行某些任务；

一般实现任务的定时运行有如下几种方式：
1）while …sleep
2）Threading.Timer
3）schedule模块
4）Apscheduler模块（推荐）

import time
def taskDetail(loopTime:int):
    currTime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    print(f"loopTime:{loopTime}-->","currTime:",currTime)

def taskMonitor():
    loopFlag = True
    startPoint = 1
    while loopFlag:
        if startPoint <=5:
            taskDetail(startPoint)
            startPoint = startPoint + 1
            time.sleep(5)
        else:
            print("startPoint达到阈值上限，退出循环")
            break


if __name__ == "__main__":
    taskMonitor()
    loopLevel = 0
    while (loopLevel <= 10):
        print(f"loopLevel:-->{loopLevel}")
        loopLevel = loopLevel + 1
        time.sleep(1)

结果输出：

D:SoftWarePythonpython.exe E:/PythonProject/pythonProject/PyQt5/TaskDetail.py
loopTime:1--> currTime: 2022-06-02 10:27:53
loopTime:2--> currTime: 2022-06-02 10:27:58
loopTime:3--> currTime: 2022-06-02 10:28:03
loopTime:4--> currTime: 2022-06-02 10:28:08
loopTime:5--> currTime: 2022-06-02 10:28:13
startPoint达到阈值上限，退出循环
loopLevel:-->0
loopLevel:-->1
loopLevel:-->2
loopLevel:-->3
loopLevel:-->4
loopLevel:-->5
loopLevel:-->6
loopLevel:-->7
loopLevel:-->8
loopLevel:-->9
loopLevel:-->10

Process finished with exit code 0

说明：该方式只能实现任务的周期运行，但是无法异步调用；

import time
from threading import Timer


def taskDetail(loopTime: int):
    currTime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    print(f"loopTime:{loopTime}-->", "currTime:", currTime)


def taskMonitor():
    taskDetail(2)
    t = Timer(5, taskMonitor)
    t.start()

if __name__ == "__main__":
    taskMonitor()
    loopLevel = 0
    while (loopLevel <= 10):
        print(f"loopLevel:-->{loopLevel}")
        loopLevel = loopLevel + 1
        time.sleep(1)

输出结果：

D:SoftWarePythonpython.exe E:/PythonProject/pythonProject/PyQt5/TaskDetail.py
loopTime:2--> currTime: 2022-06-02 10:20:49
loopLevel:-->0
loopLevel:-->1
loopLevel:-->2
loopLevel:-->3
loopLevel:-->4
loopTime:2--> currTime: 2022-06-02 10:20:54
loopLevel:-->5
loopLevel:-->6
loopLevel:-->7
loopLevel:-->8
loopLevel:-->9
loopTime:2--> currTime: 2022-06-02 10:20:59
loopLevel:-->10

Process finished with exit code -1

说明：
1）使用 Threading.Timer 可以实现任务的异步调用，但是 该方法实际为一个递归调用，有一定的递归深度限制；
2）即使调整了递归深度限制，运行到达到最大CPU时，python会直接销毁程序；

• 修改递归深度

sys.setrecursionlimit(100000000)

• 安装 schedule库

pip install schedule

• 代码实现

import time
import schedule


def taskDetail(taskName: str):
    currTime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    print(f"{taskName}-->", "currTime:", currTime)


def scheduleMonitor():
    schedule.clear()
    # 创建一个按3秒间隔执行任务
    schedule.every(1).seconds.do(taskDetail("Task-A"))
    # 创建一个按2秒间隔执行任务
    schedule.every(3).seconds.do(taskDetail("Task-B"))
    while True:
        schedule.run_pending()


if __name__ == "__main__":
    scheduleMonitor()
    loopLevel = 0
    while (loopLevel <= 10):
        print(f"loopLevel:-->{loopLevel}")
        loopLevel = loopLevel + 1
        time.sleep(1)

说明：该模块的任务调用模拟没有实现；该模块需要while Ture配合使用，而且占用的CPU也比其他几种方式高。

框架手册：https://apscheduler.readthedocs.io/en/stable/userguide.html

• 安装 APScheduler库

pip install APScheduler

• 代码实现

触发器决定了任务的运行时间、运行周期和运行频率；按照不同的运行要求可以分为 date、interval、cron三类；

1）date：在指定的时间，运行一次任务；
2）interval：以固定的时间间隔持续运行作业；
3）cron：在一天中特定的时间周期性运行任务；
4）trigger-list 触发器，即组合触发器，可以将 date、interval、cron 三类触发器组合实现任务的运行；

(一) date 触发

此处 schedule.add_job 的参数 trigger配置为 date，run_date设置运行时间；
run_date 的格式可以直接以日期或时间格式，也可以以字符串形式；

• run_date=date(2022 ,6, 5)

• run_date=datetime(2022 ,6, 5, 18, 00, 00)

• run_date="2022-06-22"

• run_date="2022-06-02 15:00:10"

• 代码实现

import time
from apscheduler.schedulers.blocking import BlockingScheduler


def taskDetail(taskName: str):
    currTime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    print(f"{taskName}-->", "currTime:", currTime)


if __name__ == "__main__":
    apSchedule = BlockingScheduler()
    apSchedule.add_job(func=taskDetail, trigger='date', run_date="2022-06-02 15:00:10", args=['task-A'])

    apSchedule.start()

(二) interval 触发

1）此处 schedule.add_job 的参数 trigger配置为 interval；
2）当 start_date没有指定时，将以 datetime.now() + interval作为第一次开始运行的时间；
3）运行参数：

• weeks (int)
• days (int)
• hours (int)
• minutes (int)
• seconds (int)
• start_date (datetime|str)
• end_date (datetime|str)
• timezone (datetime.tzinfo|str)
• jitter (int|None)– 延迟作业执行的最大时间，单位秒(second)；

• 代码实现

if __name__ == "__main__":
    apSchedule = BlockingScheduler()
    apSchedule.add_job(func=taskDetail, trigger='interval',  minutes=1, seconds=5,
                       end_date="2022-06-02 16:05:00", args=['task-A'])

    apSchedule.start()

结果输出：

D:SoftWarePythonpython.exe E:/PythonProject/pythonProject/PyQt5/apscheduleDemo.py
mainThread--> currTime: 2022-06-02 16:18:03
D:SoftWarePythonlibsite-packagesapschedulerutil.py:436: PytzUsageWarning: The localize method is no longer necessary, as this time zone supports the fold attribute (PEP 495). For more details on migrating to a PEP 495-compliant implementation, see https://pytz-deprecation-shim.readthedocs.io/en/latest/migration.html
  return tzinfo.localize(dt)
task-A--> currTime: 2022-06-02 16:19:08
task-A--> currTime: 2022-06-02 16:20:13

Process finished with exit code -1

说明：此处 任务运行的时间间隔为 65s；

• 时间间隔

schedule.add_job 的参数配置中，如下几种方式等同：

• trigger='interval', seconds=65
• trigger='interval', minutes=1, seconds=5

同理超出 60s的时间间隔，可以直接以 seconds 参数设置，或者结合 seconds + minutes + hours + …+ weeks 组合设置，即其他粒度的间隔参数只是便于时间间隔的单位换算；

• 运行时间

当 start_date设置的时间小于 当前时间时，则任务实际的运行时间是 start_date + n*interval大于当前时间的最近 一次时间；
如当前时间为：2022-06-02 16:31:03，设置的 start_date="2022-06-02 16:28:00", seconds=65，则首次运行的实际时间为：2022-06-02 16:31:15；

• 周期+定时运行

当需要在某个固定时间，周期性运行某个任务时，需要配合 start_date和 seconds/minutes/...weeks或者 seconds + minutes + hours+...+weeks等的组合实现；

缺点：interval 参数无法实现不同时间粒度的运行 要求，如在每个月的某几天运行，或者每天的某几个小时运行；

(三) cron 触发

• 时间粒度

1）此处 schedule.add_job 的参数 trigger配置为 cron；
2）时间粒度参数：

• year (int|str)–- 4digit year
• month (int|str)– month (1-12)
• day (int|str)– day of month (1-31)
• week (int|str)– ISO week (1-53)
• day_of_week (int|str)– number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun)
• hour (int|str)– hour (0-23)
• minute (int|str)– minute (0-59)
• second (int|str)– second (0-59)
• start_date (datetime|str)– earliest possible date/time to trigger on (inclusive)
• end_date (datetime|str)– latest possible date/time to trigger on (inclusive)
• timezone (datetime.tzinfo|str)– time zone to use for the date/time calculations (defaults to scheduler timezone)
• jitter (int|None)– delay the job execution by seconds at mostjitter

---

3）当时间粒度参数的具体值是单个数值时，可以使用 int 类型，当是某个区间或其他序列时，使用 str 格式；

• 参数阈值

1）如下列表中的表达式适用于 年 到 秒，在一个时间粒度的参数中，阈值可以使用多个表达式的组合，以逗号分割；
2）每个时间粒度的参数设置了后，都起作用；如同时设置了 day、hour、minute、second，那么任务将在对应时间粒度的时间内运行；

英文的第几全程简写对照如下：
first / 1st，second / 2nd，third /3rd，fourth /4th，
fifth / 5th，sixth / 6th，seventh / 7th，eighth / 8th，
ninth /9th，tenth / 10th，eleventh / 11th，twelfth / 12th

---

星期的周几英文简写和数字对照如下：
mon(0)，tue(1)，wed(2)，thu(3)，fri(4)，sat(5)，sun(6)

• 代码实现

if __name__ == "__main__":
    apSchedule = BlockingScheduler()
    apSchedule.add_job(func=taskDetail,trigger="cron",day="5th thu",args=["task-A"])
    print("mainThread-->", "currTime:", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
    apSchedule.start()

说明：上述设置为 每个月的第二个周四运行；

• 装饰器调用
  代码实现：

import time
from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()


@scheduler.scheduled_job(trigger='cron', second="*/13", args=["task-A"])
def taskDetail(taskName: str):
    currTime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    print(f"{taskName}-->", "currTime:", currTime)


if __name__ == "__main__":
    print("mainThread-->", "currTime:", time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
    scheduler.start()

说明：此处的调度器变量申明需要放到 函数申请的前面，否则装饰器无法使用，变量不识别；

• 时区切换

当时区由标准时区切换为夏令时，或者由夏令时切换为标准时区时，任务调度的时间时间将提前半小时或滞后半小时，可以通过禁用夏令时来规避该问题的出现；
可以设置时区参数：timezone="UTC"；

4） 组合触发器
对于单个触发器，无法满足某些场景：
如：在每周的周二，2:00:00运行；或者 每周三的 13:00:00运行；
此时可以使用多个触发器组合来实现特定的运行要求；

组合触发器分为2类：
A：与触发器(AndTrigger)；
B：或触发器(OrTrigger)；

如下代码实现每周周四 2:00:00、7:00:00 或者 每周日 15:00:00、19:00:00任务运行；

• 或触发组合

import time
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.combining import OrTrigger
from apscheduler.triggers.cron import CronTrigger


def taskDetail(taskName: str):
    currTime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    print(f"{taskName}-->", "currTime:", currTime)


trigList = OrTrigger([CronTrigger(day_of_week="thu", hour="2,7", minute=00, second=00),
                      CronTrigger(day_of_week="sun", hour="15,19", minute=30, second=00)])

apscheduler = BlockingScheduler()
apscheduler.add_job(func=taskDetail, trigger=trigList, args=["task-A"])
apscheduler.start()

• 与触发器

import time
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.combining import AndTrigger
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger


def taskDetail(taskName: str):
    currTime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    print(f"{taskName}-->", "currTime:", currTime)


trigList = AndTrigger([CronTrigger(day_of_week="thu", hour="2,7", minute=00, second=00),
                      CronTrigger(day_of_week="sun", hour="15,19", minute=30, second=00)])

apscheduler = BlockingScheduler()
apscheduler.add_job(func=taskDetail, trigger=trigList, args=["task-A"])
apscheduler.start()

说明：此处没有实现 AndTrigger 触发逻辑；

====================================== over ==================================