python實現定時任務那些你不知道的模塊

一、使用time中的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

threading模塊中的Timer是一個非阻塞函數，比sleep稍好一點，不過依然無法叫你起床

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

Timer類第一個參數是時間間隔(單位是秒)，第二個參數是要盜用的函數名，第三個參數調用函數的參數(是一個tuple)

三、sched模塊

sched模塊是python內置的模塊，它是一個調度(延時處理機制)，每次想要定時執行某任務必須要寫一個調度

import sched
import time
from datetime import datetime

# 初始化sched模塊的 scheduler 類
# 第一個參數是一個可以返回時間戳的函數，第二個參數可以在定時未到達之前阻塞。
schedule = sched.scheduler(time.time, time.sleep)


# 被周期性調度觸發的函數
def printTime(n):
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    schedule.enter(n, 0, printTime, (n,))  # 每隔10秒執行一個printTime函數


# 默認參數5s
def main(n=5):
    # enter四個參數分別為：間隔時間、優先級（用於同時間到達的兩個事件同時執行時定序）、被調用觸發的函數，
    # 給該觸發函數的參數（tuple形式）
    schedule.enter(6, 0, printTime, (n,))  # 6秒后執行一次printTime函數
    schedule.run()


# 10s 輸出一次
main(10)

sched 使用步驟如下：

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

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

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

四、APScheduler定時框架

終於到了你要找的可以叫你起床的定時任務了

APScheduler是一個python定時任務框架，使用起來也十分方面，提供了基於日期，固定時間 間隔以及crontab類型的任務，並且可以持久化任務，兵役daemon(守護進程)方式運行應用

使用APSchduler模塊需要先進性安裝

pip3 install apscheduler

下面是實現一個在周一到周五指定的時刻執行任務

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime


# 輸出時間
def job():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))  # 周一到周五的上午十點四十三分叫我起床


# BlockingScheduler
scheduler = BlockingScheduler()
scheduler.add_job(job, 'cron', day_of_week='1-5', hour=10, minute=43)
scheduler.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

上面的列子已經滿足我們的基本使用，如果想深入了解請看下面內容：

APScheduler四個組件

APScheduler 四個組件分別為：觸發器(trigger)，作業存儲(job store)，執行器(executor)，調度器(scheduler)。

觸發器(trigger)

包含調度邏輯，每一個作業有它自己的觸發器，用於決定接下來哪一個作業會運行。除了他們自己初始配置意外，觸發器完全是無狀態的
APScheduler 有三種內建的 trigger:

date: 特定的時間點觸發
interval: 固定時間間隔觸發
cron: 在特定時間周期性地觸發

作業存儲(job store)

存儲被調度的作業，默認的作業存儲是簡單地把作業保存在內存中，其他的作業存儲是將作業保存在數據庫中。一個作業的數據講在保存在持久化作業存儲時被序列化，並在加載時被反序列化。調度器不能分享同一個作業存儲。
APScheduler 默認使用 MemoryJobStore，可以修改使用 DB 存儲方案

執行器(executor)

處理作業的運行，他們通常通過在作業中提交制定的可調用對象到一個線程或者進城池來進行。當作業完成時，執行器將會通知調度器。
最常用的 executor 有兩種：

ProcessPoolExecutor
ThreadPoolExecutor

調度器(scheduler)

通常在應用中只有一個調度器，應用的開發者通常不會直接處理作業存儲、調度器和觸發器，相反，調度器提供了處理這些的合適的接口。配置作業存儲和執行器可以在調度器中完成，例如添加、修改和移除作業。

配置調度器

APScheduler提供了許多不同的方式來配置調度器，你可以使用一個配置字典或者作為參數關鍵字的方式傳入。你也可以先創建調度器，再配置和添加作業，這樣你可以在不同的環境中得到更大的靈活性。

執行時間間隔執行一次任務

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime


def func():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))


# 定義BlockingScheduler
sched = BlockingScheduler()
sched.add_job(func, 'interval', seconds=5)  # 每個5秒執行一次func函數
sched.start()

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

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

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 func():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))


# 存儲方式
jobstores = {
    'mongo': MongoDBJobStore(collection='test', database='local', 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(func, 'interval', seconds=5, jobstore='mongo')  # 每隔5秒執行一個func函數
scheduler.start()

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

對 job 的操作

添加 job

添加job有兩種方式：

1. add_job()
2. scheduled_job()

第二種方法只適用於應用運行期間不會改變的 job，而第一種方法返回一個apscheduler.job.Job 的實例，可以用來改變或者移除 job。

from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
# 裝飾器
@sched.scheduled_job('interval', id='my_job_id', seconds=5)
def job_function():
    print("Hello World")
# 開始
sched.start()

@sched.scheduled_job() 是 Python 的裝飾器。

移除 job

移除 job 也有兩種方法：

1. remove_job()
2. job.remove()

remove_job 使用 jobID 移除
job.remove() 使用 add_job() 返回的實例

job = scheduler.add_job(myfunc, 'interval', minutes=2)
job.remove()
# id
scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
scheduler.remove_job('my_job_id')

暫停和恢復 job

暫停一個 job：

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

恢復一個 job：

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

希望你還記得 apscheduler.job.Job 是 add_job() 返回的實例

獲取 job 列表

獲得可調度 job 列表，可以使用get_jobs() 來完成，它會返回所有的 job 實例。

也可以使用print_jobs() 來輸出所有格式化的 job 列表。

修改 job

除了 jobID 之外 job 的所有屬性都可以修改，使用 apscheduler.job.Job.modify() 或者 modify_job() 修改一個 job 的屬性

job.modify(max_instances=6, name='Alternate name')
modify_job('my_job_id', trigger='cron', minute='*/5')

關閉 job

默認情況下調度器會等待所有的 job 完成后，關閉所有的調度器和作業存儲。將 wait 選項設置為 False 可以立即關閉。

scheduler.shutdown()
scheduler.shutdown(wait=False)

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)

trigger 規則

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()

cron

• year (int|str) – 4-digit year
• month (int|str) – month (1-12)
• day (int|str) – day of the (1-31)
• week (int|str) – ISO week (1-53)
• day_of_week (int|str) – number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun)
• hour (int|str) – hour (0-23)
• minute (int|str) – minute (0-59)
• second (int|str) – second (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)

表達式:

from apscheduler.schedulers.blocking import BlockingScheduler


def job_function():
    print("Hello World")


# BlockingScheduler
sched = BlockingScheduler()
# Schedules job_function to be run on the third Friday
# of June, July, August, November and December at 00:00, 01:00, 02:00 and 03:00
sched.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
# Runs from Monday to Friday at 5:30 (am) until 2014-05-30 00:00:00
sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2014-05-30')
sched.start()

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

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()