Python定时任务 - apscheduler

• • 前言
  • 实际场景代码
  • 原理篇
  • • 一、 四大组件⚙️
    • • 触发器 triggers ：用于设定触发任务的条件
      • 任务储存器 job stores：用于存放任务，把任务存放在内存或数据库中
      • 执行器 executors： 用于执行任务，可以设定执行模式为单线程或线程池
      • 调度器 schedulers： 把上方三个组件作为参数，通过创建调度器实例来运行
    • 二、 其他樂
    • • 1. 监听事件⏰
    • 三、 一些细节
    • • add_job彩

• 项目中进程用到的定时任务，或者后台异步周期任务，可以使用apscheduler来进行管理， 下面通过两部分说明使用方式， 首先第一部分先给出实际的项目使用代码场景，第二部分给出原理

• 下面的代码构建了一个调度器, 实现了任务调度，任务删除，任务暂停，任务管理等功能

  # 采用非阻塞式调度器进行任务的运行
  import datetime
  import json
  import os
  import threading
  import traceback
  import logging
  
  import apscheduler.jobstores.base
  from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR, EVENT_JOB_SUBMITTED, EVENT_JOB_REMOVED
  from apscheduler.schedulers.background import BackgroundScheduler
  from apscheduler.triggers.interval import IntervalTrigger
  from apscheduler.triggers.cron import CronTrigger
  from apscheduler.triggers.date import DateTrigger
  from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
  from apscheduler.util import undefined
  
  from db.sqlite.base import BaseDb
  from common.base_model import Strategy, Trigger
  import common
  from utils import parse_argv
  
  logger = logging.getLogger("HomedTools")
  
  
  class Scheduler(BaseDb):
      __table_name__ = "scheduler"
      STATE_OPEN = 1
      STATE_OFF = 0
  
      def __init__(self):
          super().__init__()
          self.ThreadPool = ThreadPoolExecutor(20)
          self.ProcessPool = ProcessPoolExecutor(10)
  
          self.table_name = self.create_table(self.__table_name__, self._create_sql)
  
          self.scheduler = BackgroundScheduler(executors={
              'default': self.ThreadPool,
              # "process": self.ProcessPool
          })
          self.trigger_map = {
              "cron": CronTrigger,
              "interval": IntervalTrigger,
              "date": DateTrigger}
          self.strategy = dict()
          self._load_strategy()  # 策略容器  key - 策略id; value - 策略对象
          self.strategy_cursor = max(self.strategy) if self.strategy else 1001
          self._start()  # 策略启动
          self.threadingLock = threading.Lock()
  
      def feature(self):
          pass
  
      @property
      def strategy_max(self):
          with common.Lock:
              self.strategy_cursor += 1
              return self.strategy_cursor
  
      @property
      def _create_sql(self):
          return ("CREATE TABLE `{}` "
                  "("
                  "f_strategy_id   INT(11)   NOT NULL DEFAULT 1101, "
                  "f_strategy_name  VARCHAr(256)   NOT NULL DEFAULT '', "
                  "f_status  INT(6)   NOT NULL DEFAULT 1, "
                  "f_model_name   VARCHAr(256)   NOT NULL DEFAULT '', "
                  "f_trigger   TEXT  NOT NULL  DEFAULT '', "
                  "f_genre   VARCHAr(66) NOT NULL, "
                  "f_run_now   INT(6) NOT NULL DEFAULT 0, "
                  "f_args   TEXT NOT NULL DEFAULT '', "
                  "f_create_time  TIMESTAMP NOT NULL  DEFAULT CURRENT_TIMESTAMP, "
                  "PRIMARY KEY (f_strategy_id)"
                  "); ")
  
      def _load_strategy(self):  # 策略加载
  
          sql = f"SELECt f_strategy_id, f_strategy_name, f_model_name, f_trigger, " 
              f"f_genre, f_status, f_run_now, f_args FROM `{self.table_name}`"
          for each in self.execute(sql=sql):
              strategy_id = each[0]
              strategy_name = each[1]
              model_name = each[2]
              trigger = json.loads(each[3])
              trigger["genre"] = each[4]
              status = each[-3]
              run_now = each[-2]
              args = each[-1]
  
              if model_name in common.modules.models:
  
                  strategy_obj = Strategy(
                      model=common.modules.models[model_name],
                      trigger=trigger,
                      strategy_id=strategy_id,
                      strategy_name=strategy_name,
                      status=status,
                      args=args
                  )
  
              else:
                  strategy_obj = Strategy(
                      model=None,
                      trigger=trigger,
                      strategy_id=strategy_id,
                      strategy_name=strategy_name,
                      status=status,
                      args=args
                  )
                  logger.warning(f"[Scheduler:_load_strategy] {model_name} is not exist!!!")
  
              self._work(strategy_obj, run_now=run_now, is_load=1)
  
          logger.info(f"[Scheduler:_load_strategy] _load_strategy_end "
                      f"all_strategy_ram={self.strategy}, "
                      f"jobs={self.scheduler.get_jobs()}")
  
      def add(self, *args):  # 添加策略
          logger.info(f"[Scheduler:add] 添加策略  args={args}")
          strategy_id, strategy_name, model_name, run_now, trigger, status, params = args
          strategy_obj = Strategy(
              model=common.modules.models[model_name],
              trigger=trigger,
              strategy_id=strategy_id,
              strategy_name=strategy_name,
              status=status,
              args=params
          )
          return self._work(strategy_obj, run_now=run_now)
  
      def update(self, strategy_id, trigger, args):  # 更改策略
          return self.modify_strategy(strategy_id, trigger, args)
  
      def no_repeat(self, model_name):
          for k, v in self.strategy.items():
              if v.model.model_name == model_name:
                  return v.strategy_id
  
      @staticmethod
      def job_id(strategy_id, pre="strategy"):
          return f"{pre}_{strategy_id}"
  
      @property
      def all_jobs(self):
          return [i.id for i in self.scheduler.get_jobs()]
  
      def modify_strategy(self, strategy_id, trigger, args):
          try:
              if strategy_id in self.strategy:
                  strategy = self.strategy[strategy_id]
  
                  logger.info(f"[Scheduler:modify_strategy] strategy_trigger={strategy.trigger_fields}")
                  job_id = self.job_id(strategy_id)
                  trigger_base = Trigger(trigger=trigger)
                  genre, trigger = trigger_base.trigger_genre, trigger_base.trigger_fields
  
                  trigger_obj = self.trigger_map[genre](**trigger)
                  with self.threadingLock:
                      # 改动暂停策略执行
                      logger.info(f"[Scheduler:modify_strategy] scheduler_get_jobs={self.scheduler.get_jobs()}")
                      try:
                          self.scheduler.pause_job(job_id)
                      except apscheduler.jobstores.base.JobLookupError:  # job_id 不在 scheduler, 已经执行完毕了 需要重新 add进去
                          logger.info(f"[Scheduler:modify_strategy] no_have_job {job_id}, new one")
                          if strategy.trigger_genre != "date":
                              res = self.scheduler.add_job(func=strategy.model.job,
                                                           trigger=trigger_obj,
                                                           id=job_id,
                                                           args=args if args else None,
                                                           kwargs={"debug": 0},
                                                           next_run_time=undefined)
                          else:
                              res = self.scheduler.add_job(func=strategy.model.job,
                                                           trigger=trigger_obj,
                                                           id=job_id,
                                                           args=args if args else None,
                                                           kwargs={"debug": 0})
                          logger.info(f"[Scheduler:modify_strategy] strategy_work_success, "
                                      f"res={res}, call={strategy.model.job}"
                                      f" jobs={[i.func_ref for i in self.scheduler.get_jobs()]}")
                      else:
                          self.strategy[strategy_id].status = self.STATE_OFF
                          # 改动策略执行条件
  
                          self.scheduler.modify_job(job_id=job_id, trigger=trigger_obj, args=args)
  
                          self.scheduler.resume_job(job_id)
                          self.strategy[strategy_id].status = self.STATE_OPEN
                      return 1
              else:
  
                  return 0
          except Exception:
              logger.error(f"[Scheduler:modify_strategy] n {traceback.format_exc()}")
              return -1
  
      def pause(self, strategy_id=None):  # 暂停 运行中的脚本不会干预
          if strategy_id is None:
  
              pause_flag = self.scheduler.pause()
          else:
  
              pause_flag = self.scheduler.pause_job(self.job_id(strategy_id))
          dump_flag = self._set_status(strategy_id=strategy_id, status=0)
  
          if all((pause_flag, dump_flag)):
              return 1
          else:
              return 0
  
      def _set_status(self, strategy_id=None, status=None):
  
          sql = f"UPDATE `{self.table_name}` SET f_status={status}"
          if strategy_id:
              self.strategy[strategy_id].status = status
              dump_flag = self.execute(sql=sql+f" WHERe f_strategy_id={strategy_id}")
          else:
              for strategy_id, strategy_obj in self.strategy.items():
                  strategy_obj.status = status
              dump_flag = self.execute(sql=sql)
          return dump_flag
  
      def resume(self, strategy_id=None):  # 恢复
          if strategy_id is None:
  
              pause_flag = self.scheduler.resume()
          else:
              pause_flag = self.scheduler.resume_job(job_id=self.job_id(strategy_id))
          dump_flag = self._set_status(strategy_id=strategy_id, status=1)
          if all((pause_flag, dump_flag)):
              return 1
          else:
              return 0
  
      def remove(self, strategy_id):  # 删除策略 运行中的脚本不会干预
  
          if strategy_id is None:
  
              self.scheduler.remove_all_jobs()
          else:
              if strategy_id not in self.strategy:
                  logger.warning(f"[Scheduler:remove] strategy_no_exist {strategy_id}")
                  return 0
              try:
                  job_id = self.job_id(strategy_id)
                  self.scheduler.remove_job(job_id=job_id)
  
              except apscheduler.jobstores.base.JobLookupError:
  
                  logger.warning(f"[Scheduler:remove] strategy_no_in_job {strategy_id}")
  
              self.strategy.pop(strategy_id)
          return 1
  
      def event_driven(self):  # 事件驱动
          def job_error(event):  #
  
              logger.error(f"[Scheduler:event_driven:job_error] job {event.job_id}, execute error  n {event.traceback} "
                           f"nn {event.exception}")
  
          def job_start(event):  # 开始执行
              logger.info(f"[Scheduler:event_driven:job_start] job {event.job_id}")
  
          def job_end(event):  # 结束执行
              logger.info(f"[Scheduler:event_driven:job_end] job {event.job_id}")
  
          def job_delete(event):  # 删除任务
              logger.info(f"[Scheduler:event_driven:job_delete] job {event.job_id}")
  
          self.scheduler.add_listener(job_start, EVENT_JOB_SUBMITTED)  # 任务开始
          self.scheduler.add_listener(job_error, EVENT_JOB_ERROR)  # 任务结束后 监听完成和异常时间
          self.scheduler.add_listener(job_end, EVENT_JOB_EXECUTED)  # 任务结束
          self.scheduler.add_listener(job_delete, EVENT_JOB_REMOVED)  # 任务结束
  
      def _start(self):
          self.event_driven()
          self.scheduler.start()
          logger.info('[[Scheduler:_start] 策略启动')
  
      def _work(self, strategy: Strategy, run_now, is_load=0):  # 开始策略执行
  
          if strategy.model is None:
              self.strategy[strategy.strategy_id] = strategy
              return 1
          if strategy.status == self.STATE_OFF:
              logger.warning(f"[Scheduler:_work] strategy_is_off {strategy.strategy_name}")
              self.strategy[strategy.strategy_id] = strategy
              return 1
          if strategy.trigger_genre is not None:
              if is_load and strategy.trigger_genre == "date":
                  logger.warning(f"[Scheduler:_work] run_once_no_load {strategy.strategy_name} ")
                  self.strategy[strategy.strategy_id] = strategy
                  return 1
              trigger = self.trigger_map[strategy.trigger_genre](
                  **strategy.trigger_fields)
              args = strategy.args
              logger.info(f"[Scheduler:_work] "
                          f"strategy_parse_success, id {strategy.strategy_id},"
                          f" name {strategy.strategy_name},"
                          f" genre {strategy.trigger_genre}"
                          f" fields_value {strategy.trigger_fields}"
                          f" args={args}"
                          )
  
              if strategy.trigger_genre != "date":
                  res = self.scheduler.add_job(func=strategy.model.job,
                                               trigger=trigger,
                                               id=self.job_id(strategy.strategy_id),
                                               args=args if args else None,
                                               kwargs={"debug": 0},
                                               next_run_time=datetime.datetime.now() if run_now else undefined)
              else:
                  res = self.scheduler.add_job(func=strategy.model.job,
                                               trigger=trigger,
                                               id=self.job_id(strategy.strategy_id),
                                               args=args if args else None,
                                               kwargs={"debug": 0}
                                               )
              logger.info(f"[Scheduler:_work] strategy_work_success, run_now={run_now} res={res}, call={strategy.model.job}"
                          f" jobs={[i.func_ref for i in self.scheduler.get_jobs()]}")
              self.strategy[strategy.strategy_id] = strategy
              return 1
          else:
              logger.error(f'[Scheduler:_work] strategy_work_error, run_now={run_now}, id {strategy.strategy_id}, name {strategy.strategy_name},'
                           f' genre {strategy.trigger_genre}')
              return 0	
  

• date 日期：触发任务运行的具体日期
• interval 间隔：触发任务运行的时间间隔
• cron 周期：触发任务运行的周期

• AsyncIOExecutor
• GeventExecutor
• ThreadPoolExecutor
• ProcessPoolExecutor
• TornadoExecutor
• TwistedExecutor

• BlockingScheduler 阻塞式调度器：适用于只跑调度器的程序。
• BackgroundScheduler 后台调度器：适用于非阻塞的情况，调度器会在后台独立运行。
• AsyncIOScheduler AsyncIO调度器，适用于应用使用AsnycIO的情况。
• GeventScheduler Gevent调度器，适用于应用通过Gevent的情况。
• TornadoScheduler Tornado调度器，适用于构建Tornado应用。
• TwistedScheduler Twisted调度器，适用于构建Twisted应用。
• QtScheduler Qt调度器，适用于构建Qt应用。

EVENT_ALL = (EVENT_SCHEDULER_STARTED | 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)

• max_instances

  max_instances (int) – maximum number of concurrently running instances allowed for this job

  如果程序的运行时间是 10秒， 但是调度设置的是 5s 执行一次， 那此参数就代表此种场景的并发数

• next_run_time
  我们可以 next_run_time=datetime.datetime.now() 指定该任务在启动初就开始执行