在Python開發(fā)過程中我們經(jīng)常需要執(zhí)行定時任務(wù)帐我，而此類任務(wù)我們通常有如下選項：
- 自己造輪子
- 使用schedule庫
- 使用Celery定時任務(wù)
- 使用APScheduler
自己造輪子實現(xiàn)婶溯，最大的優(yōu)勢就是靈活性，調(diào)試方便排嫌，對于某些特定系統(tǒng)也許也是一種選擇宇立，不過對于大多數(shù)應(yīng)用來說慨飘，我們應(yīng)當盡可能地使用開源的成熟的方案。下面對后三種方案分別討論：
使用schedule庫
schedule庫是一個輕量級的定時任務(wù)方案国撵，優(yōu)勢是使用簡單，也不需要做什么配置玻墅；缺點是無法動態(tài)添加任務(wù)介牙，也無法將任務(wù)持久化。
安裝
pip install schedule
使用
import schedule
import time
def job():
print("I'm working...")
schedule.every(10).minutes.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)
schedule.every(5).to(10).minutes.do(job)
schedule.every().monday.do(job)
schedule.every().wednesday.at("13:15").do(job)
while True:
schedule.run_pending()
time.sleep(1)
使用Celery
Celery在Python領(lǐng)域可謂大名鼎鼎澳厢，我們通常將Celery作為一個任務(wù)隊列來使用环础，不過Celery也同時提供了定時任務(wù)功能。通常剩拢，當我們的解決方案中已經(jīng)在使用Celery的時候可以考慮同時使用其定時任務(wù)功能线得，但是Celery無法在Flask這樣的系統(tǒng)中動態(tài)添加定時任務(wù)（在Django中有相應(yīng)的插件可以實現(xiàn)動態(tài)添加任務(wù)），而且如果對于不使用Celery的項目徐伐，單獨為定時任務(wù)搭建Celery顯得過于重量級了贯钩。(搭建Celery比較麻煩，還需要配置諸如RabbitMQ之類消息分發(fā)程序)办素。
Celery安裝在此不再贅述角雷，大家可以參考官網(wǎng)的資料
使用
Celery雖然無法動態(tài)添加定時任務(wù)，但是可以在程序固定位置添加定時任務(wù)性穿，如下：
from celery import Celery
from celery.schedules import crontab
app = Celery()
# 此處on_after_configure裝飾符意味著當Celery app配置完成之后調(diào)用該hook函數(shù)
@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
# Calls test('hello') every 10 seconds.
sender.add_periodic_task(10.0, test.s('hello'), name='add every 10')
# Calls test('world') every 30 seconds
sender.add_periodic_task(30.0, test.s('world'), expires=10)
# Executes every Monday morning at 7:30 a.m.
sender.add_periodic_task(
crontab(hour=7, minute=30, day_of_week=1),
test.s('Happy Mondays!'),
)
@app.task
def test(arg):
print(arg)
- 這里調(diào)用
add_periodic_task用于添加一個定時任務(wù)勺三，相當于在Celery config文件中的beat_schedule設(shè)置項中添加了一項，如下：app.conf.beat_schedule = { 'add-every-30-seconds': { 'task': 'tasks.add', 'schedule': 30.0, 'args': (16, 16) }, } - 在
add_periodic_task中指定job function時需要用.s()來調(diào)用
使用APScheduler
筆者認為APScheduler是在實際項目最好用的一個工具庫需曾。它不僅可以讓我們在程序中動態(tài)添加和刪除我們的定時任務(wù)檩咱，還支持持久化揭措，且其持久化方案支持很多形式，包括(Memory, MongoDB, SQLAlchemy, Redis, RethinkDB, ZooKeeper), 也可以非常好與一些Python framework集成(包括asyncio, gevent, Tornado, Twisted, Qt). 筆者所在的項目使用的是Flask框架刻蚯，也有相應(yīng)的插件可以供我們直接使用绊含。
但是筆者沒有使用插件，而是直接將APScheduler集成于項目代碼中炊汹。
初始化scheduler
# 可以在初始化Flask的時候調(diào)用躬充，并將返回的scheduler賦給app
def init_scheduler():
# 這里用于持久化的設(shè)置，代碼中演示使用MongoDB
# client用于設(shè)置你自己的MongoDB的handler, 即MongoClient對象
jobstores = {
'default': MongoDBJobStore(client=your_db_handler, collection="schedule_job")
}
executors = {
'default': ThreadPoolExecutor(20)
}
job_defaults = {
'coalesce': False,
'max_instances': 5
}
# 這里使用BackgroundScheduler即可
scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)
# 注意這里一定要調(diào)用start啟動scheduler
scheduler.start()
return scheduler
添加定時任務(wù)
APScheduler將定時任務(wù)分為三種：
- interval: 比如每隔5分鐘執(zhí)行一次任務(wù)
- cron: 比如每天早上5點執(zhí)行一次任務(wù)
- date: 比如在2018年5月5日執(zhí)行一次任務(wù)
我們以添加cron job為例：
def test_job(name):
print "hello, %s" % name
def add_daily_job(name):
exec_time = datetime.now() + timedelta(minutes=2)
hour = exec_time.strftime("%H")
minute = exec_time.strftime("%M")
# 這里要選擇'cron'
# 另外讨便，job_id可以根據(jù)你自己的情況設(shè)定充甚，其會被用于remove_job
current_app.scheduler.add_job(
test_job, 'cron', hour=hour, minute=minute,
args=[name], id=job_id)
刪除定時任務(wù)
通過在add_job時使用的job_id可以刪除對應(yīng)的定時任務(wù)。實際上在我們添加任務(wù)的時候霸褒，APScheduler會把相應(yīng)的任務(wù)信息存儲于我們jobstore中設(shè)置的持久化存儲方案伴找，這里使用的是MongoDB，然后當刪除的時候會將相應(yīng)的任務(wù)從MongoDB中刪除废菱。
def remove_daily_job(job_id):
current_app.scheduler.remove_job(job_id)
總結(jié)：
APScheduler在實際使用過程中擁有最大的靈活性技矮，可以滿足我們的大部分定時任務(wù)的相關(guān)需求；Celery比較重量級殊轴，通常如果項目中已有Celery在使用衰倦，而且不需要動態(tài)添加定時任務(wù)時可以考慮使用；schedule非常輕量級旁理，使用簡單樊零，但是不支持任務(wù)的持久化，也無法動態(tài)添加刪除任務(wù)孽文，所以主要用于簡單的小型應(yīng)用驻襟。
