APScheduler定時任務使用

python之APScheduler模塊使用

• 介紹
• 主要模塊
• 使用

介紹

需要使用定時任務, 簡單的可以使用cron模塊完成, 考慮到window及linux的通用性, 使用APScheduler模塊實現

• 文檔地址
• windows和linux下都能使用
• 解決幾種類型的定時任務
  • 一次性延時運行: 在設定的時間點運行一次任務
  • 循環定時運行: cron型的定時運行,可指定開始和結束時間
  • 基於間隔的執行: 以一定間隔運行作業,可指定開始和結束時間

主要模塊

• 觸發器[triggers]
• 作業店[job stores]
• 執行者[executors]
• 調度器[schedulers]

使用

• 安裝

  pip install apscheduler

• 創建調度器

  from datetime import datetime
  import json
  
  from apscheduler.schedulers.background import BackgroundScheduler
  from apscheduler.jobstores.mongodb import MongoDBJobStore
  from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
  from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor  
  from apscheduler.jobstores.redis import RedisJobStore  
  import redis
  import pickle  
  connect_args = {
      'host' :'127.0.0.1',
      'port': 6379,
      'password':'123456'
  }
  jobstores = {
      'default': RedisJobStore(db=14,
                          jobs_key='apscheduler.jobs',
                          run_times_key='apscheduler.run_times',
                          pickle_protocol=pickle.HIGHEST_PROTOCOL,
                          **connect_args)
  }  
  executors = {
      'default': ThreadPoolExecutor(20),
      'processpool': ProcessPoolExecutor(5)
  }
  job_defaults = {
      'coalesce': False,
      'max_instances': 3
  }  
  scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors,  job_defaults=job_defaults, timezone="Asia/Shanghai")
  scheduler.start()
  

• 創建觸發函數，添加到調度器

  def my_func():
      """
      定時任務的測試函數
      """
      print('我正在執行定時任務!')
  
  
  job = scheduler.add_job(my_func, 'interval', minutes=1, id='my_job_id')
  

• 任務可以，獲取列表，暫停，修改(更新),建議使用時設定id