python定时任务contrib_python定时任务的实现
背景
目前所在的項(xiàng)目組需要經(jīng)常執(zhí)行一些定時(shí)任務(wù)，之前都是用 Node.JS 的?cron來實(shí)現(xiàn) schedule job。可是這次需要連接不同的 DB，而且實(shí)現(xiàn)的邏輯也有些許不同，于是選擇使用 Python 的定時(shí)器。
Python 實(shí)現(xiàn)定時(shí)任務(wù)
循環(huán) sleep
這種方式最簡(jiǎn)單，在循環(huán)里面放入要執(zhí)行的任務(wù)，然后 sleep 一段時(shí)間再執(zhí)行
1
2
3
4
5
6
7
8
9
from datetime import datetime
import time
# 每n秒執(zhí)行一次
def timer(n):
while True:
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
time.sleep(n)
# 5s
timer(5)
這個(gè)方法的缺點(diǎn)是，只能執(zhí)行固定間隔時(shí)間的任務(wù)，如果有定時(shí)任務(wù)就無法完成，比如早上六點(diǎn)半喊我起床。并且 sleep 是一個(gè)阻塞函數(shù)，也就是說 sleep 這一段時(shí)間，啥都不能做。
threading模塊中的Timer
threading 模塊中的 Timer 是一個(gè)非阻塞函數(shù)，比 sleep 稍好一點(diǎn)，不過依然無法喊我起床。
1
2
3
4
5
6
7
8
9
from datetime import datetime
from threading import Timer
# 打印時(shí)間函數(shù)
def printTime(inc):
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
t = Timer(inc, printTime, (inc,))
t.start()
# 5s
printTime(5)
Timer 函數(shù)第一個(gè)參數(shù)是時(shí)間間隔(單位是秒)，第二個(gè)參數(shù)是要調(diào)用的函數(shù)名，第三個(gè)參數(shù)是調(diào)用函數(shù)的參數(shù)(tuple)
使用sched模塊
sched 模塊是 Python 內(nèi)置的模塊，它是一個(gè)調(diào)度(延時(shí)處理機(jī)制)，每次想要定時(shí)執(zhí)行某任務(wù)都必須寫入一個(gè)調(diào)度。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import sched
import time
from datetime import datetime
# 初始化sched模塊的 scheduler 類
# 第一個(gè)參數(shù)是一個(gè)可以返回時(shí)間戳的函數(shù)，第二個(gè)參數(shù)可以在定時(shí)未到達(dá)之前阻塞。
schedule = sched.scheduler(time.time, time.sleep)
# 被周期性調(diào)度觸發(fā)的函數(shù)
def printTime(inc):
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
schedule.enter(inc, 0, printTime, (inc,))
# 默認(rèn)參數(shù)60s
def main(inc=60):
# enter四個(gè)參數(shù)分別為：間隔事件、優(yōu)先級(jí)(用于同時(shí)間到達(dá)的兩個(gè)事件同時(shí)執(zhí)行時(shí)定序)、被調(diào)用觸發(fā)的函數(shù)，
# 給該觸發(fā)函數(shù)的參數(shù)(tuple形式)
schedule.enter(0, 0, printTime, (inc,))
schedule.run()
# 10s 輸出一次
main(10)
sched 使用步驟如下：
(1)生成調(diào)度器：
s = sched.scheduler(time.time,time.sleep)
第一個(gè)參數(shù)是一個(gè)可以返回時(shí)間戳的函數(shù)，第二個(gè)參數(shù)可以在定時(shí)未到達(dá)之前阻塞。
(2)加入調(diào)度事件
其實(shí)有 enter、enterabs 等等，我們以 enter 為例子。
s.enter(x1,x2,x3,x4)
四個(gè)參數(shù)分別為：間隔事件、優(yōu)先級(jí)(用于同時(shí)間到達(dá)的兩個(gè)事件同時(shí)執(zhí)行時(shí)定序)、被調(diào)用觸發(fā)的函數(shù)，給觸發(fā)函數(shù)的參數(shù)(注意：一定要以 tuple 給，如果只有一個(gè)參數(shù)就(xx,))
(3)運(yùn)行
s.run()
注意 sched 模塊不是循環(huán)的，一次調(diào)度被執(zhí)行后就 Over 了，如果想再執(zhí)行，請(qǐng)?jiān)俅?enter
APScheduler定時(shí)框架
終于找到了可以每天定時(shí)喊我起床的方式了
APScheduler是一個(gè) Python 定時(shí)任務(wù)框架，使用起來十分方便。提供了基于日期、固定時(shí)間間隔以及 crontab 類型的任務(wù)，并且可以持久化任務(wù)、并以 daemon 方式運(yùn)行應(yīng)用。
使用 APScheduler 需要安裝
1
$ pip install apscheduler
首先來看一個(gè)周一到周五每天早上6點(diǎn)半喊我起床的例子
1
2
3
4
5
6
7
8
9
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
# 輸出時(shí)間
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=6, minute=30)
scheduler.start()
代碼中的 BlockingScheduler 是什么呢？
BlockingScheduler 是 APScheduler 中的調(diào)度器，APScheduler 中有兩種常用的調(diào)度器，BlockingScheduler 和 BackgroundScheduler，當(dāng)調(diào)度器是應(yīng)用中唯一要運(yùn)行的任務(wù)時(shí)，使用 BlockingSchedule，如果希望調(diào)度器在后臺(tái)執(zhí)行，使用 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四個(gè)組件
APScheduler 四個(gè)組件分別為：觸發(fā)器(trigger)，作業(yè)存儲(chǔ)(job store)，執(zhí)行器(executor)，調(diào)度器(scheduler)。
觸發(fā)器(trigger)
包含調(diào)度邏輯，每一個(gè)作業(yè)有它自己的觸發(fā)器，用于決定接下來哪一個(gè)作業(yè)會(huì)運(yùn)行。除了他們自己初始配置意外，觸發(fā)器完全是無狀態(tài)的
APScheduler 有三種內(nèi)建的 trigger:
date: 特定的時(shí)間點(diǎn)觸發(fā)
interval: 固定時(shí)間間隔觸發(fā)
cron: 在特定時(shí)間周期性地觸發(fā)
作業(yè)存儲(chǔ)(job store)
存儲(chǔ)被調(diào)度的作業(yè)，默認(rèn)的作業(yè)存儲(chǔ)是簡(jiǎn)單地把作業(yè)保存在內(nèi)存中，其他的作業(yè)存儲(chǔ)是將作業(yè)保存在數(shù)據(jù)庫(kù)中。一個(gè)作業(yè)的數(shù)據(jù)講在保存在持久化作業(yè)存儲(chǔ)時(shí)被序列化，并在加載時(shí)被反序列化。調(diào)度器不能分享同一個(gè)作業(yè)存儲(chǔ)。
APScheduler 默認(rèn)使用 MemoryJobStore，可以修改使用 DB 存儲(chǔ)方案
執(zhí)行器(executor)
處理作業(yè)的運(yùn)行，他們通常通過在作業(yè)中提交制定的可調(diào)用對(duì)象到一個(gè)線程或者進(jìn)城池來進(jìn)行。當(dāng)作業(yè)完成時(shí)，執(zhí)行器將會(huì)通知調(diào)度器。
最常用的 executor 有兩種：
ProcessPoolExecutor
ThreadPoolExecutor
調(diào)度器(scheduler)
通常在應(yīng)用中只有一個(gè)調(diào)度器，應(yīng)用的開發(fā)者通常不會(huì)直接處理作業(yè)存儲(chǔ)、調(diào)度器和觸發(fā)器，相反，調(diào)度器提供了處理這些的合適的接口。配置作業(yè)存儲(chǔ)和執(zhí)行器可以在調(diào)度器中完成，例如添加、修改和移除作業(yè)。
配置調(diào)度器
APScheduler提供了許多不同的方式來配置調(diào)度器，你可以使用一個(gè)配置字典或者作為參數(shù)關(guān)鍵字的方式傳入。你也可以先創(chuàng)建調(diào)度器，再配置和添加作業(yè)，這樣你可以在不同的環(huán)境中得到更大的靈活性。
下面來看一個(gè)簡(jiǎn)單的 BlockingScheduler 例子
1
2
3
4
5
6
7
8
9
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(job, 'interval', seconds=5)
sched.start()
上述代碼創(chuàng)建了一個(gè) BlockingScheduler，并使用默認(rèn)內(nèi)存存儲(chǔ)和默認(rèn)執(zhí)行器。(默認(rèn)選項(xiàng)分別是 MemoryJobStore 和 ThreadPoolExecutor，其中線程池的最大線程數(shù)為10)。配置完成后使用 start() 方法來啟動(dòng)。
如果想要顯式設(shè)置 job store(使用mongo存儲(chǔ))和 executor 可以這樣寫：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
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 參數(shù)
host = '127.0.0.1'
port = 27017
client = MongoClient(host, port)
# 輸出時(shí)間
def job():
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
# 存儲(chǔ)方式
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()
在運(yùn)行程序5秒后，第一次輸出時(shí)間。
在 MongoDB 中可以看到 job 的狀態(tài)
對(duì) job 的操作
添加 job
添加job有兩種方式：
add_job()
scheduled_job()
第二種方法只適用于應(yīng)用運(yùn)行期間不會(huì)改變的 job，而第一種方法返回一個(gè)apscheduler.job.Job?的實(shí)例，可以用來改變或者移除 job。
1
2
3
4
5
6
7
8
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 也有兩種方法：
remove_job()
job.remove()
remove_job 使用 jobID 移除
job.remove() 使用 add_job() 返回的實(shí)例
1
2
3
4
5
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')
暫停和恢復(fù) job
暫停一個(gè) job：
1
2
apscheduler.job.Job.pause()
apscheduler.schedulers.base.BaseScheduler.pause_job()
恢復(fù)一個(gè) job：
1
2
apscheduler.job.Job.resume()
apscheduler.schedulers.base.BaseScheduler.resume_job()
希望你還記得 apscheduler.job.Job 是 add_job() 返回的實(shí)例
獲取 job 列表
獲得可調(diào)度 job 列表，可以使用get_jobs()?來完成，它會(huì)返回所有的 job 實(shí)例。
也可以使用print_jobs()?來輸出所有格式化的 job 列表。
修改 job
除了 jobID 之外 job 的所有屬性都可以修改，使用 apscheduler.job.Job.modify() 或者 modify_job() 修改一個(gè) job 的屬性
1
2
job.modify(max_instances=6, name='Alternate name')
modify_job('my_job_id', trigger='cron', minute='*/5')
關(guān)閉 job
默認(rèn)情況下調(diào)度器會(huì)等待所有的 job 完成后，關(guān)閉所有的調(diào)度器和作業(yè)存儲(chǔ)。將 wait 選項(xiàng)設(shè)置為 False 可以立即關(guān)閉。
1
2
scheduler.shutdown()
scheduler.shutdown(wait=False)
scheduler 事件
scheduler 可以添加事件監(jiān)聽器，并在特殊的時(shí)間觸發(fā)。
1
2
3
4
5
6
7
def my_listener(event):
if event.exception:
print('The job crashed :(')
else:
print('The job worked :)')
# 添加監(jiān)聽器
scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
trigger 規(guī)則
最基本的一種調(diào)度，作業(yè)只會(huì)執(zhí)行一次。它的參數(shù)如下：
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
1
2
3
4
5
6
7
8
9
10
11
12
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()
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)
表達(dá)式:
1
2
3
4
5
6
7
8
9
10
11
12
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()
參數(shù)：
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
1
2
3
4
5
6
7
8
9
10
11
12
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()
總結(jié)
以上是生活随笔為你收集整理的python定时任务contrib_python定时任务的实现的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 写一个公用的gpio口驱动
- 下一篇： python操作crontab定时任务