Python 實現定時任務

參考鏈接

一、循環 sleep

這種方式最簡單，在循環里面放入要執行的任務，然后 sleep 一段時間再執行

from datetime import datetime
import time
# 每n秒執行一次
def timer(n):
    while True:
        print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
        time.sleep(n)
# 5s
timer(5)

這個方法的缺點是：只能執行固定間隔時間的任務，如果有定時任務就無法完成，比如早上六點半喊我起床。並且 sleep 是一個阻塞函數，也就是說 sleep 這一段時間，啥都不能做。對服務器性能的損耗。

二、threading模塊中的Timer

Timer 函數第一個參數是時間間隔（單位是秒，只有秒），第二個參數是要調用的函數名，第三個參數是調用函數的參數(tuple)。

from datetime import datetime
from threading import Timer
# 打印時間函數
def printTime(inc):
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    t = Timer(inc, printTime, (inc,))
    t.start()
# 5s
printTime(5)

三、使用sched模塊

sched 模塊是 Python 內置的模塊，它是一個調度（延時處理機制），每次想要定時執行某任務都必須寫入一個調度。

import sched
import time
from datetime import datetime
# 初始化sched模塊的 scheduler 類
# 第一個參數是一個可以返回時間戳的函數，第二個參數可以在定時未到達之前阻塞。
schedule = sched.scheduler(time.time, time.sleep)
# 被周期性調度觸發的函數
def printTime(inc):
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    schedule.enter(inc, 0, printTime, (inc,))
# 默認參數60s
def main(inc=60):
    # enter四個參數分別為：間隔事件、優先級（用於同時間到達的兩個事件同時執行時定序）、被調用觸發的函數，
    # 給該觸發函數的參數（tuple形式）
    schedule.enter(0, 0, printTime, (inc,))
    schedule.run()
# 10s 輸出一次
main(10)

sched 使用步驟如下：

• 生成調度器：
  s = sched.scheduler(time.time,time.sleep)
  第一個參數是一個可以返回時間戳的函數，第二個參數可以在定時未到達之前阻塞。

• 加入調度事件
  其實有 enter、enterabs 等等，我們以 enter 為例子。
  s.enter(x1,x2,x3,x4)
  四個參數分別為：間隔事件、優先級（用於同時間到達的兩個事件同時執行時定序）、被調用觸發的函數，給觸發函數的參數（注意：一定要以 tuple 給，如果只有一個參數就(xx,)）

• 運行
  s.run()
  注意 sched 模塊不是循環的，一次調度被執行后就 Over 了，如果想再執行，請再次 enter

四、定時任務框架APScheduler學習詳解

參考鏈接

一、APScheduler簡介

• 背景

  在平常的工作中幾乎有一半的功能模塊都需要定時任務來推動，例如項目中有一個定時統計程序，定時爬出網站的URL程序，定時檢測釣魚網站的程序等等，都涉及到了關於定時任務的問題，第一時間想到的是利用time模塊的time.sleep()方法使程序休眠來達到定時任務的目的，雖然這樣也可以，但是總覺得不是那么的專業，^_所以就找到了python的定時任務模塊APScheduler：

• APScheduler基於Quartz的一個Python定時任務框架，實現了Quartz的所有功能，使用起來十分方便。提供了基於日期、固定時間間隔以及crontab類型的任務，並且可以持久化任務， 並以 daemon 方式運行應用。基於這些功能，我們可以很方便的實現一個python定時任務系統。

二、安裝

使用 APScheduler 需要安裝

• 利用pip進行安裝

pip install apscheduler

• 源碼安裝（https://pypi.python.org/pypi/APScheduler/）

  python setup.py install
  

三、APScheduler有四種組成部分：

• 觸發器(trigger)包含調度邏輯，每一個任務有它自己的觸發器，用於決定接下來哪一個任務會運行。除了他們自己初始配置意外，觸發器完全是無狀態的。APScheduler 有三種內建的 trigger:
  • date: 特定的時間點觸發
  • interval: 固定時間間隔觸發
  • cron: 在特定時間周期性地觸發
• 任務存儲(job store)存儲被調度的任務，默認的任務存儲是簡單地把任務保存在內存中，其他的任務存儲是將任務保存在數據庫中。一個任務的數據講在保存在持久化任務存儲時被序列化，並在加載時被反序列化。調度器不能分享同一個任務存儲。
• 執行器(executor)處理任務的運行，他們通常通過在任務中提交制定的可調用對象到一個線程或者進城池來進行。當任務完成時，執行器將會通知調度器。最常用的 executor 有兩種：
  • ProcessPoolExecutor
  • ThreadPoolExecutor
• 調度器(scheduler)是其他的組成部分。你通常在應用只有一個調度器，應用的開發者通常不會直接處理任務存儲、調度器和觸發器，相反，調度器提供了處理這些的合適的接口。配置任務存儲和執行器可以在調度器中完成，例如添加、修改和移除任務。

四、簡單應用

import time
from apscheduler.schedulers.blocking import BlockingScheduler
 
def my_job():
    print time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
 
sched = BlockingScheduler()
# 每隔5s執行一次my_job函數，輸出當前時間信息
sched.add_job(my_job, 'interval', seconds=5)
# 時間： 周一到周五每天早上6點半, 執行my_job
sched.add_job(my_job, 'cron', day_of_week='1-5', hour=6, minute=30)
sched.start()

# 注釋
代碼中的 BlockingScheduler 是什么呢？

BlockingScheduler是APScheduler中的調度器，APScheduler 中有兩種常用的調度器，BlockingScheduler 和 BackgroundScheduler，當調度器是應用中唯一要運行的任務時，使用 BlockingSchedule，如果希望調度器在后台執行，使用 BackgroundScheduler。

BlockingScheduler: use when the scheduler is the only thing running in your process
BackgroundScheduler: use when you’re not using any of the frameworks below, and want the scheduler to run in the background inside your application
AsyncIOScheduler: use if your application uses the asyncio module
GeventScheduler: use if your application uses gevent
TornadoScheduler: use if you’re building a Tornado application
TwistedScheduler: use if you’re building a Twisted application
QtScheduler: use if you’re building a Qt application

五、任務操作

1.添加任務

上面是通過add_job()來添加任務，另外還有一種方式是通過scheduled_job()修飾器來修飾函數

import time
from apscheduler.schedulers.blocking import BlockingScheduler
 
sched = BlockingScheduler()
 
@sched.scheduled_job('interval', seconds=5)
def my_job():
    print time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
 
sched.start()

上述代碼創建了一個 BlockingScheduler，並使用默認內存存儲和默認執行器。(默認選項分別是 MemoryJobStore 和 ThreadPoolExecutor，其中線程池的最大線程數為10)。配置完成后使用 start() 方法來啟動。

如果想要顯式設置 job store(使用mongo存儲)和 executor 可以這樣寫：

# 在運行程序5秒后，第一次輸出時間。
# 在 MongoDB 中可以看到 job 的狀態

from datetime import datetime
from pymongo import MongoClient
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.jobstores.memory import MemoryJobStore
from apscheduler.jobstores.mongodb import MongoDBJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
# MongoDB 參數
host = '127.0.0.1'
port = 27017
client = MongoClient(host, port)
# 輸出時間
def job():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
# 存儲方式
jobstores = {
    'mongo': MongoDBJobStore(collection='job', database='test', client=client),
    'default': MemoryJobStore()
}
executors = {
    'default': ThreadPoolExecutor(10),
    'processpool': ProcessPoolExecutor(3)
}
job_defaults = {
    'coalesce': False,
    'max_instances': 3
}
scheduler = BlockingScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults)
scheduler.add_job(job, 'interval', seconds=5, jobstore='mongo')
scheduler.start()

2.刪除任務

job = scheduler.add_job(myfunc, 'interval', minutes=2)
job.remove()
#如果有多個任務序列的話可以給每個任務設置ID號，可以根據ID號選擇清除對象，且remove放到start前才有效
sched.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
sched.remove_job('my_job_id')

3. 暫停和恢復任務

• 暫停任務

  apsched.job.Job.pause()
  apsched.schedulers.base.BaseScheduler.pause_job()
  

• 恢復任務

  apsched.job.Job.resume()
  apsched.schedulers.base.BaseScheduler.resume_job()
  

4.獲取job列表

獲得調度任務的列表，可以使用get_jobs()來完成，它會返回所有的job實例。或者使用print_jobs()來輸出所有格式化的任務列表。也可以利用get_job(任務ID)獲取指定任務的任務列表

job = sched.add_job(my_job, 'interval', seconds=2 ,id='123')
print sched.get_job(job_id='123')
print sched.get_jobs()

5.關閉調度器

默認情況下調度器會等待所有正在運行的任務完成后，關閉所有的調度器和任務存儲。如果你不想等待，可以將wait選項設置為False。

sched.shutdown()
sched.shutdown(wait=False)

6.scheduler 事件

scheduler 可以添加事件監聽器，並在特殊的時間觸發。

def my_listener(event):
    if event.exception:
        print('The job crashed :(')
    else:
        print('The job worked :)')
# 添加監聽器
scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)

六、任務運行控制

add_job的第二個參數是trigger，它管理着任務的調度方式。它可以為date, interval或者cron。對於不同的trigger，對應的參數也相同。

1. 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) -（表示時區取值）

• 案例

  #表示2020年7月22日17時19分07秒執行該程序
  sched.add_job(my_job, 'cron', year=2020,month = 7,day = 22,hour = 17,minute = 19,second = 7)
   
  #表示任務在6,7,8,11,12月份的第三個星期五的00:00,01:00,02:00,03:00 執行該程序
  sched.add_job(my_job, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
   
  #表示從星期一到星期五5:30（AM）直到2020-08-30 00:00:00
  sched.add_job(my_job(), 'cron', day_of_week='mon-fri', hour=5, minute=30,end_date='2020-08-30')
   
  #表示每5秒執行該程序一次，相當於interval 間隔調度中seconds = 5
  sched.add_job(my_job, 'cron',second = '*/5')
  

2.interval 間隔調度（每隔多久執行）

weeks (int) – number of weeks to wait
days (int) – number of days to wait
hours (int) – number of hours to wait
minutes (int) – number of minutes to wait
seconds (int) – number of seconds to wait
start_date (datetime|str) – starting point for the interval calculation
end_date (datetime|str) – latest possible date/time to trigger on
timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations

• 案例

  #表示每隔3天17時19分07秒執行一次任務
  sched.add_job(my_job, 'interval',days  = 3,hours = 17,minutes = 19,seconds = 7)
  
  from datetime import datetime
  from apscheduler.schedulers.blocking import BlockingScheduler
  
  def job_function():
      print("Hello World")
      
  # BlockingScheduler
  sched = BlockingScheduler()
  
  # Schedule job_function to be called every two hours
  sched.add_job(job_function, 'interval', hours=2)
  
  # The same as before, but starts on 2010-10-10 at 9:30 and stops on 2014-06-15 at 11:00
  sched.add_job(job_function, 'interval', hours=2, start_date='2010-10-10 09:30:00', end_date='2014-06-15 11:00:00')
  
  sched.start()
  

3.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

• 案例

  from datetime import date
  from apscheduler.schedulers.blocking import BlockingScheduler
  sched = BlockingScheduler()
  def my_job(text):
      print(text)
  # The job will be executed on November 6th, 2009
  sched.add_job(my_job, 'date', run_date=date(2009, 11, 6), args=['text'])
  sched.add_job(my_job, 'date', run_date=datetime(2009, 11, 6, 16, 30, 5), args=['text'])
  sched.add_job(my_job, 'date', run_date='2009-11-06 16:30:05', args=['text'])
  # The 'date' trigger and datetime.now() as run_date are implicit
  sched.add_job(my_job, args=['text'])
  sched.start()