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

上一篇 / 下一篇  2021-09-27 09:22:59

  6. 利用任务框架APScheduler实现定时任务
  APScheduler(advanceded python scheduler)基于Quartz的一个Python定时任务框架,实现了Quartz的所有功能,使用起来十分方便。提供了基于日期、固定时间间隔以及crontab类型的任务,并且可以持久化任务。基于这些功能,我们可以很方便的实现一个Python定时任务系统。
  搜索公众号Linux中文社区猿后台回复“私房菜”,获取一份惊喜礼包。
  它有以下三个特点:
  ·类似于 Liunx Cron 的调度程序(可选的开始/结束时间)
  · 基于时间间隔的执行调度(周期性调度,可选的开始/结束时间)
  · 一次性执行任务(在设定的日期/时间运行一次任务)
  APScheduler有四种组成部分:
  · 触发器(trigger) 包含调度逻辑,每一个作业有它自己的触发器,用于决定接下来哪一个作业会运行。除了他们自己初始配置意外,触发器完全是无状态的。
  · 作业存储(job store) 存储被调度的作业,默认的作业存储是简单地把作业保存在内存中,其他的作业存储是将作业保存在数据库中。一个作业的数据讲在保存在持久化作业存储时被序列化,并在加载时被反序列化。调度器不能分享同一个作业存储。
  · 执行器(executor) 处理作业的运行,他们通常通过在作业中提交制定的可调用对象到一个线程或者进城池来进行。当作业完成时,执行器将会通知调度器。
  · 调度器(scheduler) 是其他的组成部分。你通常在应用只有一个调度器,应用的开发者通常不会直接处理作业存储、调度器和触发器,相反,调度器提供了处理这些的合适的接口。配置作业存储和执行器可以在调度器中完成,例如添加、修改和移除作业。通过配置executor、jobstore、trigger,使用线程池(ThreadPoolExecutor默认值20)或进程池(ProcessPoolExecutor 默认值5)并且默认最多3个(max_instances)任务实例同时运行,实现对job的增删改查等调度控制。
  示例代码:
  from apscheduler.schedulers.blocking import BlockingScheduler  
  from datetime import datetime  
  # 输出时间  
  def job():  
      print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))  
  # BlockingScheduler  
  sched = BlockingScheduler()  
  sched.add_job(my_job, 'interval', seconds=5, id='my_job_id')  
  sched.start() 
  6.1 Job 作业
  Job作为APScheduler最小执行单位。创建Job时指定执行的函数,函数中所需参数,Job执行时的一些设置信息。
  构建说明:
  ·id:指定作业的唯一ID
  · name:指定作业的名字
  · trigger:apscheduler定义的触发器,用于确定Job的执行时间,根据设置的trigger规则,计算得到下次执行此job的时间, 满足时将会执行
  · executor:apscheduler定义的执行器,job创建时设置执行器的名字,根据字符串你名字到scheduler获取到执行此job的 执行器,执行job指定的函数
  · max_instances:执行此job的最大实例数,executor执行job时,根据job的id来计算执行次数,根据设置的最大实例数来确定是否可执行
  · next_run_time:Job下次的执行时间,创建Job时可以指定一个时间[datetime],不指定的话则默认根据trigger获取触发时间
  · misfire_grace_time:Job的延迟执行时间,例如Job的计划执行时间是21:00:00,但因服务重启或其他原因导致21:00:31才执行,如果设置此key为40,则该job会继续执行,否则将会丢弃此job
  · coalesce:Job是否合并执行,是一个bool值。例如scheduler停止20s后重启启动,而job的触发器设置为5s执行一次,因此此job错过了4个执行时间,如果设置为是,则会合并到一次执行,否则会逐个执行
  · func:Job执行的函数
  · args:Job执行函数需要的位置参数
  · kwargs:Job执行函数需要的关键字参数
  6.2 Trigger 触发器
  Trigger绑定到Job,在scheduler调度筛选Job时,根据触发器的规则计算出Job的触发时间,然后与当前时间比较确定此Job是否会被执行,总之就是根据trigger规则计算出下一个执行时间。
  目前APScheduler支持触发器:
  · 指定时间的DateTrigger
  · 指定间隔时间的IntervalTrigger
  · 像Linux的crontab一样的CronTrigger。
  触发器参数:date
  date定时,作业只执行一次。
  · run_date (datetime|str) – the date/time to run the job at
  · timezone (datetime.tzinfo|str) – time zone for run_date if it doesn’t have one already 
  sched.add_job(my_job, 'date', run_date=date(2009, 11, 6), args=['text']) 
  sched.add_job(my_job, 'date', run_date=datetime(2019, 7, 6, 16, 30, 5), args=['text']) 
  触发器参数:interval
  interval间隔调度
  ·weeks (int) – 间隔几周
  · days (int) – 间隔几天
  · hours (int) – 间隔几小时
  · minutes (int) – 间隔几分钟
  · seconds (int) – 间隔多少秒
  · start_date (datetime|str) – 开始日期
  · end_date (datetime|str) – 结束日期
  · timezone (datetime.tzinfo|str) – 时区 
  sched.add_job(job_function, 'interval', hours=2) 
  触发器参数:cron
  cron调度
  · (int|str) 表示参数既可以是int类型,也可以是str类型
  · (datetime | str) 表示参数既可以是datetime类型,也可以是str类型
  · year (int|str) – 4-digit year -(表示四位数的年份,如2008年)
  · month (int|str) – month (1-12) -(表示取值范围为1-12月)
  · day (int|str) – day of the (1-31) -(表示取值范围为1-31日)
  · week (int|str) – ISO week (1-53) -(格里历2006年12月31日可以写成2006年-W52-7(扩展形式)或2006W527(紧凑形式))
  · day_of_week (int|str) – number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun) – (表示一周中的第几天,既可以用0-6表示也可以用其英语缩写表示)
  · hour (int|str) – hour (0-23) – (表示取值范围为0-23时)
  · minute (int|str) – minute (0-59) – (表示取值范围为0-59分)
  · second (int|str) – second (0-59) – (表示取值范围为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) -(表示时区取值)
  CronTrigger可用的表达式:
  # 6-8,11-12月第三个周五 00:00, 01:00, 02:00, 03:00运行  
  sched.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')  
  # 每周一到周五运行 直到2024-05-30 00:00:00  
  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:
  ·executors.asyncio:同步io,阻塞
  · executors.gevent:io多路复用,非阻塞
  · executors.pool: 线程ThreadPoolExecutor和进程ProcessPoolExecutor
  · executors.twisted:基于事件驱动
   6.4 Jobstore 作业存储
  Jobstore在scheduler中初始化,另外也可通过scheduler的add_jobstore动态添加Jobstore。每个jobstore都会绑定一个alias,scheduler在Add Job时,根据指定的jobstore在scheduler中找到相应的jobstore,并将job添加到jobstore中。作业存储器决定任务的保存方式, 默认存储在内存中(MemoryJobStore),重启后就没有了。APScheduler支持的任务存储器有:
  ·  jobstores.memory:内存
  ·  jobstores.mongodb:存储在mongodb
  ·  jobstores.redis:存储在redis
  ·  jobstores.rethinkdb:存储在rethinkdb
  ·  jobstores.sqlalchemy:支持sqlalchemy的数据库如mysql,sqlite等
  ·  jobstores.zookeeper:zookeeper
  不同的任务存储器可以在调度器的配置中进行配置(见调度器)
   6.5 Event 事件
  Event是APScheduler在进行某些操作时触发相应的事件,用户可以自定义一些函数来监听这些事件,当触发某些Event时,做一些具体的操作。常见的比如。Job执行异常事件 EVENT_JOB_ERROR。Job执行时间错过事件 EVENT_JOB_MISSED。
  目前APScheduler定义的Event:
  · EVENT_SCHEDULER_STARTED
  · EVENT_SCHEDULER_START
  · EVENT_SCHEDULER_SHUTDOWN
  · EVENT_SCHEDULER_PAUSED
  · EVENT_SCHEDULER_RESUMED
  · EVENT_EXECUTOR_ADDED
  · EVENT_EXECUTOR_REMOVED
  · EVENT_JOBSTORE_ADDED
  · EVENT_JOBSTORE_REMOVED
  · EVENT_ALL_JOBS_REMOVED
  · EVENT_JOB_ADDED
  · EVENT_JOB_REMOVED
  · EVENT_JOB_MODIFIED
  · EVENT_JOB_EXECUTED
  · EVENT_JOB_ERROR
  · EVENT_JOB_MISSED
  · EVENT_JOB_SUBMITTED
  · EVENT_JOB_MAX_INSTANCES
  Listener表示用户自定义监听的一些Event,比如当Job触发了EVENT_JOB_MISSED事件时可以根据需求做一些其他处理。
   6.6 调度器
  Scheduler是APScheduler的核心,所有相关组件通过其定义。scheduler启动之后,将开始按照配置的任务进行调度。除了依据所有定义Job的trigger生成的将要调度时间唤醒调度之外。当发生Job信息变更时也会触发调度。
  APScheduler支持的调度器方式如下,比较常用的为BlockingScheduler和BackgroundScheduler
  · BlockingScheduler:适用于调度程序是进程中唯一运行的进程,调用start函数会阻塞当前线程,不能立即返回。
  · BackgroundScheduler:适用于调度程序在应用程序的后台运行,调用start后主线程不会阻塞。
  · AsyncIOScheduler:适用于使用了asyncio模块的应用程序。
  · GeventScheduler:适用于使用gevent模块的应用程序。
  · TwistedScheduler:适用于构建Twisted的应用程序。
  · QtScheduler:适用于构建Qt的应用程序。
   6.7 Scheduler的工作流程
  Scheduler添加job流程:
  Scheduler调度流程:
  7. 使用分布式消息系统Celery实现定时任务
  Celery是一个简单,灵活,可靠的分布式系统,用于处理大量消息,同时为操作提供维护此类系统所需的工具, 也可用于任务调度。Celery 的配置比较麻烦,如果你只是需要一个轻量级的调度工具,Celery 不会是一个好选择。
  Celery 是一个强大的分布式任务队列,它可以让任务的执行完全脱离主程序,甚至可以被分配到其他主机上运行。我们通常使用它来实现异步任务(async task)和定时任务(crontab)。异步任务比如是发送邮件、或者文件上传, 图像处理等等一些比较耗时的操作 ,定时任务是需要在特定时间执行的任务。
  需要注意,celery本身并不具备任务的存储功能,在调度任务的时候肯定是要把任务存起来的,因此在使用celery的时候还需要搭配一些具备存储、访问功能的工具,比如:消息队列、Redis缓存、数据库等。官方推荐的是消息队列RabbitMQ,有些时候使用Redis也是不错的选择。
  它的架构组成如下图:
  Celery架构,它采用典型的生产者-消费者模式,主要由以下部分组成:
  · Celery Beat,任务调度器,Beat进程会读取配置文件的内容,周期性地将配置中到期需要执行的任务发送给任务队列。
  · Producer:需要在队列中进行的任务,一般由用户、触发器或其他操作将任务入队,然后交由workers进行处理。调用了Celery提供的API、函数或者装饰器而产生任务并交给任务队列处理的都是任务生产者。
  · Broker,即消息中间件,在这指任务队列本身,Celery扮演生产者和消费者的角色,brokers就是生产者和消费者存放/获取产品的地方(队列)。
  · Celery Worker,执行任务的消费者,从队列中取出任务并执行。通常会在多台服务器运行多个消费者来提高执行效率。
  · Result Backend:任务处理完后保存状态信息和结果,以供查询。Celery默认已支持Redis、RabbitMQ、MongoDB、Django ORM、SQLAlchemy等方式。
  实际应用中,用户从Web前端发起一个请求,我们只需要将请求所要处理的任务丢入任务队列broker中,由空闲的worker去处理任务即可,处理的结果会暂存在后台数据库backend中。我们可以在一台机器或多台机器上同时起多个worker进程来实现分布式地并行处理任务。
  Celery定时任务实例:
  · Python Celery & RabbitMQ Tutorial
  · 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实现,可以完成各种任务实现:
  · BashOperator – 执行 bash 命令或脚本。
  · SSHOperator – 执行远程 bash 命令或脚本(原理同 paramiko 模块)。
  · PythonOperator – 执行 Python 函数。
  · EmailOperator – 发送 Email。
  · HTTPOperator – 发送一个 HTTP 请求。
  · MySqlOperator, SqliteOperator, PostgresOperator, MsSqlOperator, OracleOperator, JdbcOperator, 等,执行 SQL 任务。
  · DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator…
  除了以上这些 Operators 还可以方便的自定义 Operators 满足个性化的任务需求。
  一些情况下,我们需要根据执行结果执行不同的任务,这样工作流会产生分支。如:
  这种需求可以使用BranchPythonOperator来实现。
  8.1 Airflow 产生的背景
  通常,在一个运维系统,数据分析系统,或测试系统等大型系统中,我们会有各种各样的依赖需求。包括但不限于:
  · 时间依赖:任务需要等待某一个时间点触发。
  · 外部系统依赖:任务依赖外部系统需要调用接口去访问。
  · 任务间依赖:任务 A 需要在任务 B 完成后启动,两个任务互相间会产生影响。
  · 资源环境依赖:任务消耗资源非常多, 或者只能在特定的机器上执行。
  crontab 可以很好地处理定时执行任务的需求,但仅能管理时间上的依赖。Airflow 的核心概念 DAG(有向无环图)—— 来表现工作流。
  · Airflow 是一种 WMS,即:它将任务以及它们的依赖看作代码,按照那些计划规范任务执行,并在实际工作进程之间分发需执行的任务。
  · Airflow 提供了一个用于显示当前活动任务和过去任务状态的优秀 UI,并允许用户手动管理任务的执行和状态。
  · Airflow 中的工作流是具有方向性依赖的任务集合。
  · DAG 中的每个节点都是一个任务,DAG 中的边表示的是任务之间的依赖(强制为有向无环,因此不会出现循环依赖,从而导致无限执行循环)。
  8.2 Airflow 核心概念
  · DAGs:即有向无环图(Directed Acyclic Graph),将所有需要运行的tasks按照依赖关系组织起来,描述的是所有tasks执行顺序。
  · Operators:可以简单理解为一个class,描述了DAG中某个的task具体要做的事。其中,airflow内置了很多operators,如BashOperator 执行一个bash 命令,PythonOperator 调用任意的Python 函数,EmailOperator 用于发送邮件,HTTPOperator 用于发送HTTP请求, SqlOperator 用于执行SQL命令等等,同时,用户可以自定义Operator,这给用户提供了极大的便利性。
  · Tasks:Task 是 Operator的一个实例,也就是DAGs中的一个node。
  · Task Instance:task的一次运行。Web 界面中可以看到task instance 有自己的状态,包括”running”, “success”, “failed”, “skipped”, “up for retry”等。
  · Task Relationships:DAGs中的不同Tasks之间可以有依赖关系,如 Task1 >> Task2,表明Task2依赖于Task2了。通过将DAGs和Operators结合起来,用户就可以创建各种复杂的 工作流(workflow)。
  8.3 Airflow 的架构
  在一个可扩展的生产环境中,Airflow 含有以下组件:
  · 元数据库:这个数据库存储有关任务状态的信息。
  · 调度器:Scheduler 是一种使用 DAG 定义结合元数据中的任务状态来决定哪些任务需要被执行以及任务执行优先级的过程。调度器通常作为服务运行。
  · 执行器:Executor 是一个消息队列进程,它被绑定到调度器中,用于确定实际执行每个任务计划的工作进程。有不同类型的执行器,每个执行器都使用一个指定工作进程的类来执行任务。例如,LocalExecutor 使用与调度器进程在同一台机器上运行的并行进程执行任务。其他像 CeleryExecutor 的执行器使用存在于独立的工作机器集群中的工作进程执行任务。
  · Workers:这些是实际执行任务逻辑的进程,由正在使用的执行器确定。
  Worker的具体实现由配置文件中的executor来指定,airflow支持多种Executor:
  · SequentialExecutor: 单进程顺序执行,一般只用来测试
  · LocalExecutor: 本地多进程执行
  · CeleryExecutor: 使用Celery进行分布式任务调度
  · DaskExecutor:使用Dask进行分布式任务调度
  · KubernetesExecutor: 1.10.0新增, 创建临时POD执行每次任务
  生产环境一般使用CeleryExecutor和KubernetesExecutor。
  使用CeleryExecutor的架构如图:
  使用KubernetesExecutor的架构如图:


TAG: 软件开发 Python

 

评分:0

我来说两句

Open Toolbar