python定时任务方法总结

总结如下：

①用time的sleep方法
②用threading的Timer方法
③用sched模块
④用schedule
⑤用 Timeloop 库运行定时任务
⑥用apscheduler的BlockingScheduler方法

01 time模块

这是最低级的方法，任务还需要配合着while来使用。缺点就是无法执行不同时间间隔的任务。

import time
def task():
    print("定时任务执行中...")
while True:
    task()
    time.sleep(60)  # 休眠1min

02threading模块

threading能与方法配合使用，且能多线程执行任务，缺点是时间周期与函数执行的时间需要统计，没办法做到定时触发。

from threading import Timer
def func():
  print('Hello world')
  Timer(5,func).start()
func()

03使用sched模块
Python的sched模块提供了一个事件调度器，可以在指定的时间执行函数。这是一个更灵活的解决方案，可以安排多个任务

import time
import sched
s = sched.scheduler(time.time, time.sleep)
def task(sc):
    print("定时任务执行中...")
s.enter(3600, 1, task, (s,))
s.run()

enter(delay, priority, action, argument)，安排一个事件来延迟 delay 个时间单位。

cancel(event)：从队列中删除事件。如果事件不是当前队列中的事件，则该方法将跑出一个 ValueError。

run()：运行所有预定的事件。这个函数将等待（使用传递给构造函数的 delayfunc() 函数），然后执行事件，直到不再有预定的事件。

04schedule
schedule是一个Python库，它提供了一种非常简单的方式来执行定时任务。它的语法直观易懂，适用于各种定时任务需求。

import schedule
import time
def mtask():
    print("定时任务执行中...")
schedule.every(1).hour.do(mtask)
while True:
    schedule.run_pending()
    time.sleep(1)

05用 Timeloop 库运行定时任务
是一个库，可用于运行多周期任务。这是一个简单的库，它使用 decorator 模式在线程中运行标记函数

import time
from timeloop import Timeloop
from datetime import timedelta
tl=Timeloop()
@tl.job(interval=timedelta(seconds=2))
def sample_job_every_2s():
    print("2s job current time : {}".format(time.ctime()))
if __name__=="__main__":
    tl.start(block=True)

06apscheduler框架

apscheduler比较全面，是强大的定时任务框架，可设置开始时间，结束时间，间隔时间。而且任务执行时间可以不计算，如果你的任务执行就需要耗费几秒，间隔执行虽然是对的，但是没办法做到每次的间隔都一样。apscheduler解决了这方面的烦恼。

from apscheduler.schedulers.blocking import BlockingScheduler
def func(x):
  print(x)
sched = BlockingScheduler()
#间隔打印Hello world
sched.add_job(func, 'interval', start_date= '2023-04-02 00:00:00', seconds=300, args=['Hello', ])
sched.add_job(func, 'interval', start_date= '2023-04-02 00:00:01', seconds=300, args=['world', ])

0401使用Cron表达式进行任务调度
Cron表达式是一种灵活而强大的调度方式，可以实现各种复杂的任务调度需求。APScheduler提供了CronTrigger触发器，可以方便地使用Cron表达式进行任务调度。下面是一个使用Cron表达式的示例：

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger

# 创建调度器对象
scheduler = BlockingScheduler()
# 定义任务函数
def task_func():
    print("执行任务...")

# 创建Cron触发器
trigger = CronTrigger.from_crontab('0 0 12 * * ?')
#意：移除问号'?'，因为APScheduler不支持Quartz格式的cron表达式  
# 添加定时任务,函数对象传递,
scheduler.add_job(task_func, trigger)
# 启动调度器
scheduler.start()

在这个示例中，我们使用了CronTrigger触发器，并通过from_crontab方法传入Cron表达式。这里的Cron表达式是一个字符串，用于描述任务的执行时间规则。你可以根据Cron表达式的语法，灵活地定义任务的执行时间。

标准的cron表达式通常包含5个字段，分别代表：
分钟（0-59）
小时（0-23）
日期（1-31）
月份（1-12 或者 JAN-DEC）
星期（0-7 或者 SUN-SAT，其中0和7都代表周日）

0402任务调度与多线程

在实际应用中，有些任务可能会耗时较长，为了不阻塞调度器的正常运行，我们可以将这些耗时任务放在单独的线程中执行。下面是一个使用多线程执行任务的示例：

from apscheduler.schedulers.background import BackgroundScheduler
import threading

# 创建调度器对象
scheduler = BackgroundScheduler()
# 定义任务函数
def task_func():
    print("执行任务...")
# 执行耗时任务
    threading.Thread(target=long_running_task).start()
# 添加定时任务
scheduler.add_job(task_func, 'interval', seconds=10)

# 启动调度器
scheduler.start()

结语

本文介绍了使用Python进行任务调度和自动化的基本思路和实现细节。通过合理利用Python的任务调度和自动化工具，我们可以提高工作效率，减少重复性工作的负担。希望本文能够帮助你更好地利用Python进行任务调度和自动化，提高工作效率。