文章详情

登录

短信预约-IT技能 免费直播动态提醒

选择考试省份

  • 北京
  • 上海
  • 天津
  • 重庆
  • 河北
  • 山东
  • 辽宁
  • 黑龙江
  • 吉林
  • 甘肃
  • 青海
  • 河南
  • 江苏
  • 湖北
  • 湖南
  • 江西
  • 浙江
  • 广东
  • 云南
  • 福建
  • 海南
  • 山西
  • 四川
  • 陕西
  • 贵州
  • 安徽
  • 广西
  • 内蒙
  • 西藏
  • 新疆
  • 宁夏
  • 兵团

请输入下面的图形验证码

提交验证

短信预约提醒成功

8 种 Python 定时任务的解决方案

2024-12-02 20:34

关注

最近我整理了一下 Python 定时任务的实现方式，内容较长，建议收藏后学习，梳理不易，有所收获，点赞支持。

我们开始学习吧！

# 目录

# 1. 利用while True: + sleep()实现定时任务

位于 time 模块中的 sleep(secs) 函数，可以实现令当前执行的线程暂停 secs 秒后再继续执行。所谓暂停，即令当前线程进入阻塞状态，当达到 sleep() 函数规定的时间后，再由阻塞状态转为就绪状态，等待 CPU 调度。

基于这样的特性我们可以通过while死循环+sleep()的方式实现简单的定时任务。

代码示例： 

  1. import datetime  
  2. import time  
  3. def time_printer():  
  4.     now = datetime.datetime.now()  
  5.     ts = now.strftime('%Y-%m-%d %H:%M:%S')  
  6.     print('do func time :', ts)  
  7. def loop_monitor():  
  8.     while True:  
  9.         time_printer()  
  10.         time.sleep(5)  # 暂停5秒  
  11. if __name__ == "__main__":  
  12.     loop_monitor() 

主要缺点：

# 2. 使用Timeloop库运行定时任务

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

示例代码： 

  1. import time  
  2. from timeloop import Timeloop  
  3. from datetime import timedelta  
  4. tl = Timeloop()  
  5. @tl.job(interval=timedelta(seconds=2))  
  6. def sample_job_every_2s():  
  7.     print "2s job current time : {}".format(time.ctime())  
  8. @tl.job(interval=timedelta(seconds=5))  
  9. def sample_job_every_5s():  
  10.     print "5s job current time : {}".format(time.ctime())  
  11. @tl.job(interval=timedelta(seconds=10))  
  12. def sample_job_every_10s():  
  13.     print "10s job current time : {}".format(time.ctime()) 

# 3. 利用threading.Timer实现定时任务

threading 模块中的 Timer 是一个非阻塞函数，比 sleep 稍好一点，timer最基本理解就是定时器，我们可以启动多个定时任务，这些定时器任务是异步执行，所以不存在等待顺序执行问题。

Timer(interval, function, args=[ ], kwargs={ })

代码示例

备注：Timer只能执行一次，这里需要循环调用，否则只能执行一次

# 4. 利用内置模块sched实现定时任务

sched模块实现了一个通用事件调度器，在调度器类使用一个延迟函数等待特定的时间，执行任务。同时支持多线程应用程序，在每个任务执行后会立刻调用延时函数，以确保其他线程也能执行。

class sched.scheduler(timefunc, delayfunc)这个类定义了调度事件的通用接口，它需要外部传入两个参数，timefunc是一个没有参数的返回时间类型数字的函数(常用使用的如time模块里面的time)，delayfunc应该是一个需要一个参数来调用、与timefunc的输出兼容、并且作用为延迟多个时间单位的函数(常用的如time模块的sleep)。

代码示例： 

  1. import datetime  
  2. import time  
  3. import sched  
  4. def time_printer():  
  5.     now = datetime.datetime.now()  
  6.     ts = now.strftime('%Y-%m-%d %H:%M:%S')  
  7.     print('do func time :', ts)  
  8.     loop_monitor()  
  9. def loop_monitor():  
  10.     s = sched.scheduler(time.time, time.sleep)  # 生成调度器  
  11.     s.enter(5, 1, time_printer, ())  
  12.     s.run()  
  13. if __name__ == "__main__":  
  14.     loop_monitor() 

scheduler对象主要方法:

个人点评：比threading.Timer更好，不需要循环调用。

# 5. 利用调度模块schedule实现定时任务

schedule是一个第三方轻量级的任务调度模块，可以按照秒，分，小时，日期或者自定义事件执行时间。schedule允许用户使用简单、人性化的语法以预定的时间间隔定期运行Python函数(或其它可调用函数)。

先来看代码，是不是不看文档就能明白什么意思？ 

  1. import schedule  
  2. import time  
  3. def job():  
  4.     print("I'm working...") 
  5. schedule.every(10).seconds.do(job)  
  6. schedule.every(10).minutes.do(job)  
  7. schedule.every().hour.do(job)  
  8. schedule.every().day.at("10:30").do(job)  
  9. schedule.every(5).to(10).minutes.do(job)  
  10. schedule.every().monday.do(job)  
  11. schedule.every().wednesday.at("13:15").do(job)  
  12. schedule.every().minute.at(":17").do(job)  
  13. while True:  
  14.     schedule.run_pending()  
  15.     time.sleep(1) 

装饰器：通过 @repeat() 装饰静态方法 

  1. import time  
  2. from schedule import every, repeat, run_pending  
  3. @repeat(every().second)  
  4. def job():  
  5.     print('working...')  
  6. while True:  
  7.     run_pending()  
  8.     time.sleep(1) 

传递参数： 

  1. import schedule  
  2. def greet(name):  
  3.     print('Hello', name)  
  4. schedule.every(2).seconds.do(greet, name='Alice')  
  5. schedule.every(4).seconds.do(greet, name='Bob')  
  6. while True:  
  7.     schedule.run_pending()

装饰器同样能传递参数： 

  1. from schedule import every, repeat, run_pending  
  2. @repeat(every().second, 'World')  
  3. @repeat(every().minute, 'Mars')  
  4. def hello(planet):  
  5.     print('Hello', planet)  
  6. while True:  
  7.     run_pending() 

取消任务： 

  1. import schedule  
  2. i = 0  
  3. def some_task():  
  4.     global i  
  5.     i += 1  
  6.     print(i)  
  7.     if i == 10:  
  8.         schedule.cancel_job(job)  
  9.         print('cancel job')  
  10.         exit(0)  
  11. job = schedule.every().second.do(some_task)  
  12. while True:  
  13.     schedule.run_pending() 

运行一次任务： 

  1. import time  
  2. import schedule  
  3. def job_that_executes_once():  
  4.     print('Hello')  
  5.     return schedule.CancelJob  
  6. schedule.every().minute.at(':34').do(job_that_executes_once)  
  7. while True:  
  8.     schedule.run_pending()  
  9.     time.sleep(1) 

根据标签检索任务： 

  1. # 检索所有任务：schedule.get_jobs()  
  2. import schedule  
  3. def greet(name):  
  4.     print('Hello {}'.format(name))  
  5. schedule.every().day.do(greet, 'Andrea').tag('daily-tasks', 'friend')  
  6. schedule.every().hour.do(greet, 'John').tag('hourly-tasks', 'friend')  
  7. schedule.every().hour.do(greet, 'Monica').tag('hourly-tasks', 'customer')  
  8. schedule.every().day.do(greet, 'Derek').tag('daily-tasks', 'guest')  
  9. friends = schedule.get_jobs('friend')  
  10. print(friends) 

根据标签取消任务： 

  1. # 取消所有任务：schedule.clear()  
  2. import schedule  
  3. def greet(name):  
  4.     print('Hello {}'.format(name))  
  5.     if name == 'Cancel':  
  6.         schedule.clear('second-tasks')  
  7.         print('cancel second-tasks')  
  8. schedule.every().second.do(greet, 'Andrea').tag('second-tasks', 'friend')  
  9. schedule.every().second.do(greet, 'John').tag('second-tasks', 'friend')  
  10. schedule.every().hour.do(greet, 'Monica').tag('hourly-tasks', 'customer')  
  11. schedule.every(5).seconds.do(greet, 'Cancel').tag('daily-tasks', 'guest')  
  12. while True:  
  13.     schedule.run_pending() 

运行任务到某时间： 

  1. import schedule  
  2. from datetime import datetime, timedelta, time  
  3. def job():  
  4.     print('working...')  
  5. schedule.every().second.until('23:59').do(job)  # 今天23:59停止  
  6. schedule.every().second.until('2030-01-01 18:30').do(job)  # 2030-01-01 18:30停止  
  7. schedule.every().second.until(timedelta(hours=8)).do(job)  # 8小时后停止  
  8. schedule.every().second.until(time(23, 59, 59)).do(job)  # 今天23:59:59停止  
  9. schedule.every().second.until(datetime(2030, 1, 1, 18, 30, 0)).do(job)  # 2030-01-01 18:30停止  
  10. while True:  
  11.     schedule.run_pending() 

马上运行所有任务（主要用于测试）： 

  1. import schedule  
  2. def job():  
  3.     print('working...')  
  4. def job1():  
  5.     print('Hello...')  
  6. schedule.every().monday.at('12:40').do(job)  
  7. schedule.every().tuesday.at('16:40').do(job1)  
  8. schedule.run_all()  
  9. schedule.run_all(delay_seconds=3)  # 任务间延迟3秒 

并行运行：使用 Python 内置队列实现： 

  1. import threading  
  2. import time  
  3. import schedule  
  4. def job1():  
  5.     print("I'm running on thread %s" % threading.current_thread())  
  6. def job2():  
  7.     print("I'm running on thread %s" % threading.current_thread())  
  8. def job3():  
  9.     print("I'm running on thread %s" % threading.current_thread())  
  10. def run_threaded(job_func):  
  11.     job_thread = threading.Thread(target=job_func)  
  12.     job_thread.start()  
  13. schedule.every(10).seconds.do(run_threaded, job1)  
  14. schedule.every(10).seconds.do(run_threaded, job2)  
  15. schedule.every(10).seconds.do(run_threaded, job3) 
  16. while True:  
  17.     schedule.run_pending()  
  18.     time.sleep(1) 

# 6. 利用任务框架APScheduler实现定时任务

APScheduler（advanceded python scheduler）基于Quartz的一个Python定时任务框架，实现了Quartz的所有功能，使用起来十分方便。提供了基于日期、固定时间间隔以及crontab类型的任务，并且可以持久化任务。基于这些功能，我们可以很方便的实现一个Python定时任务系统。

搜索公众号Linux中文社区猿后台回复“私房菜”，获取一份惊喜礼包。

它有以下三个特点：

APScheduler有四种组成部分：

示例代码： 

  1. from apscheduler.schedulers.blocking import BlockingScheduler  
  2. from datetime import datetime  
  3. # 输出时间  
  4. def job():  
  5.     print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))  
  6. # BlockingScheduler  
  7. sched = BlockingScheduler()  
  8. sched.add_job(my_job, 'interval', seconds=5, id='my_job_id')  
  9. sched.start() 

 6.1 Job 作业

Job作为APScheduler最小执行单位。创建Job时指定执行的函数，函数中所需参数，Job执行时的一些设置信息。

构建说明：

 6.2 Trigger 触发器

Trigger绑定到Job，在scheduler调度筛选Job时，根据触发器的规则计算出Job的触发时间，然后与当前时间比较确定此Job是否会被执行，总之就是根据trigger规则计算出下一个执行时间。

目前APScheduler支持触发器：

触发器参数：date

date定时，作业只执行一次。

  1. sched.add_job(my_job, 'date', run_date=date(2009, 11, 6), args=['text']) 
  2. sched.add_job(my_job, 'date', run_date=datetime(2019, 7, 6, 16, 30, 5), args=['text']) 

触发器参数：interval

interval间隔调度

  1. sched.add_job(job_function, 'interval', hours=2) 

触发器参数：cron

cron调度

CronTrigger可用的表达式： 

  1. # 6-8,11-12月第三个周五 00:00, 01:00, 02:00, 03:00运行  
  2. sched.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')  
  3. # 每周一到周五运行 直到2024-05-30 00:00:00  
  4. sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2024-05-30' 

 6.3 Executor 执行器

Executor在scheduler中初始化，另外也可通过scheduler的add_executor动态添加Executor。每个executor都会绑定一个alias，这个作为唯一标识绑定到Job，在实际执行时会根据Job绑定的executor找到实际的执行器对象，然后根据执行器对象执行Job。

Executor的种类会根据不同的调度来选择，如果选择AsyncIO作为调度的库，那么选择AsyncIOExecutor，如果选择tornado作为调度的库，选择TornadoExecutor，如果选择启动进程作为调度，选择ThreadPoolExecutor或者ProcessPoolExecutor都可以。

Executor的选择需要根据实际的scheduler来选择不同的执行器。目前APScheduler支持的Executor：

 6.4 Jobstore 作业存储

Jobstore在scheduler中初始化，另外也可通过scheduler的add_jobstore动态添加Jobstore。每个jobstore都会绑定一个alias，scheduler在Add Job时，根据指定的jobstore在scheduler中找到相应的jobstore，并将job添加到jobstore中。作业存储器决定任务的保存方式， 默认存储在内存中（MemoryJobStore），重启后就没有了。APScheduler支持的任务存储器有：

不同的任务存储器可以在调度器的配置中进行配置（见调度器）

 6.5 Event 事件

Event是APScheduler在进行某些操作时触发相应的事件，用户可以自定义一些函数来监听这些事件，当触发某些Event时，做一些具体的操作。常见的比如。Job执行异常事件 EVENT_JOB_ERROR。Job执行时间错过事件 EVENT_JOB_MISSED。

目前APScheduler定义的Event：

Listener表示用户自定义监听的一些Event，比如当Job触发了EVENT_JOB_MISSED事件时可以根据需求做一些其他处理。

 6.6 调度器

Scheduler是APScheduler的核心，所有相关组件通过其定义。scheduler启动之后，将开始按照配置的任务进行调度。除了依据所有定义Job的trigger生成的将要调度时间唤醒调度之外。当发生Job信息变更时也会触发调度。

APScheduler支持的调度器方式如下，比较常用的为BlockingScheduler和BackgroundScheduler

 6.7 Scheduler的工作流程

Scheduler添加job流程：

Scheduler调度流程：

# 7. 使用分布式消息系统Celery实现定时任务

Celery是一个简单，灵活，可靠的分布式系统，用于处理大量消息，同时为操作提供维护此类系统所需的工具, 也可用于任务调度。Celery 的配置比较麻烦，如果你只是需要一个轻量级的调度工具，Celery 不会是一个好选择。

Celery 是一个强大的分布式任务队列，它可以让任务的执行完全脱离主程序，甚至可以被分配到其他主机上运行。我们通常使用它来实现异步任务（async task）和定时任务（crontab）。异步任务比如是发送邮件、或者文件上传, 图像处理等等一些比较耗时的操作 ，定时任务是需要在特定时间执行的任务。

需要注意，celery本身并不具备任务的存储功能，在调度任务的时候肯定是要把任务存起来的，因此在使用celery的时候还需要搭配一些具备存储、访问功能的工具，比如：消息队列、Redis缓存、数据库等。官方推荐的是消息队列RabbitMQ，有些时候使用Redis也是不错的选择。

它的架构组成如下图：

Celery架构，它采用典型的生产者-消费者模式，主要由以下部分组成：

实际应用中，用户从Web前端发起一个请求，我们只需要将请求所要处理的任务丢入任务队列broker中，由空闲的worker去处理任务即可，处理的结果会暂存在后台数据库backend中。我们可以在一台机器或多台机器上同时起多个worker进程来实现分布式地并行处理任务。

Celery定时任务实例：

# 8. 使用数据流工具Apache Airflow实现定时任务

Apache Airflow 是Airbnb开源的一款数据流程工具，目前是Apache孵化项目。以非常灵活的方式来支持数据的ETL过程，同时还支持非常多的插件来完成诸如HDFS监控、邮件通知等功能。Airflow支持单机和分布式两种模式，支持Master-Slave模式，支持Mesos等资源调度，有非常好的扩展性。被大量公司采用。

Airflow使用Python开发，它通过DAGs(Directed Acyclic Graph, 有向无环图)来表达一个工作流中所要执行的任务，以及任务之间的关系和依赖。比如，如下的工作流中，任务T1执行完成，T2和T3才能开始执行，T2和T3都执行完成，T4才能开始执行。

Airflow提供了各种Operator实现，可以完成各种任务实现：

除了以上这些 Operators 还可以方便的自定义 Operators 满足个性化的任务需求。

一些情况下，我们需要根据执行结果执行不同的任务，这样工作流会产生分支。如:

这种需求可以使用BranchPythonOperator来实现。

 8.1 Airflow 产生的背景

通常，在一个运维系统，数据分析系统，或测试系统等大型系统中，我们会有各种各样的依赖需求。包括但不限于：

crontab 可以很好地处理定时执行任务的需求，但仅能管理时间上的依赖。Airflow 的核心概念 DAG（有向无环图）—— 来表现工作流。

 8.2 Airflow 核心概念

 8.3 Airflow 的架构

在一个可扩展的生产环境中，Airflow 含有以下组件：

Worker的具体实现由配置文件中的executor来指定，airflow支持多种Executor:

生产环境一般使用CeleryExecutor和KubernetesExecutor。

使用CeleryExecutor的架构如图:

使用KubernetesExecutor的架构如图:

其它参考：

 

来源：恋习Python内容投诉

免责声明：

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的，并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据，供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题

  • 2024年上半年信息系统项目管理师第二批次真题及答案解析（完整版）

    难度     813人已做
    查看

  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看

  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看

  • 2024年上半年软考高项第一、二批次真题考点汇总（完整版）

    难度     435人已做
    查看

  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机

8 种 Python 定时任务的解决方案

后端开发Python命令2024-12-02

python实现定时任务的8种方式详解

后端开发2023-09-09

Python实现定时任务的八种方案详解

后端开发2024-04-02

Python 实现定时任务的九种方案

后端开发Python定时任务2024-11-30

Python 4种实现定时任务的方案

后端开发2024-04-02

Python 实现定时任务的八种方案！

后端开发Python命令2024-12-02

Python定时任务实现方案

后端开发2024-04-02

python 实现定时任务的四种方式

后端开发2024-04-02

Java处理延时任务的常用几种解决方案

后端开发2024-04-02

五分钟搞定，实现定时任务的五种方案！

后端开发定时任务任务调度2024-12-01

php如何实现定时任务,php定时任务方法,最佳解决方案,php自动任务处理

后端开发2023-09-04

@Scheduled注解不能同时执行多个定时任务的解决方案

后端开发2024-04-02

java开发主流定时任务解决方案全横评详解

后端开发2024-04-02

R语言 UTF-8各种问题的解决方案

后端开发2024-04-02

Python爬虫定时计划任务的几种常见方法

后端开发Python爬虫2024-12-03

Spring Boot 实现定时任务的多种方式详解

后端开发SpringBoot定时任务2024-11-29

使用Redis实现延时任务的解决方案

后端开发2024-04-02

Java 实现定时任务的三种方法

后端开发2024-04-02

让Python程序定时执行的8种方法整理

后端开发2023-01-30

Java处理延时任务的解决方案有哪些

后端开发2023-06-30
位置：首页-资讯-后端开发
咦！没有更多了？去看看其它编程学习网 内容吧
首页课程
资料下载问答资讯