在Python開發過程中我們經常需要執行定時任務，而此類任務我們通常有如下選項：
自己造輪子
使用schedule庫
使用Celery定時任務
使用APScheduler
自己造輪子實作，最大的優勢就是靈活性，調試友善，對于某些特定系統也許也是一種選擇，不過對于大多數應用來說，我們應當盡可能地使用開源的成熟的方案。下面對後三種方案分别讨論：
使用schedule庫
schedule庫是一個輕量級的定時任務方案，優勢是使用簡單，也不需要做什麼配置；缺點是無法動态添加任務，也無法将任務持久化。
安裝
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領域可謂大名鼎鼎，我們通常将Celery作為一個任務隊列來使用，不過Celery也同時提供了定時任務功能。通常，當我們的解決方案中已經在使用Celery的時候可以考慮同時使用其定時任務功能，但是Celery無法在Flask這樣的系統中動态添加定時任務(在Django中有相應的插件可以實作動态添加任務)，而且如果對于不使用Celery的項目，單獨為定時任務搭建Celery顯得過于重量級了。(搭建Celery比較麻煩，還需要配置諸如RabbitMQ之類消息分發程式)。
Celery安裝在此不再贅述，大家可以參考官網的資料
使用
Celery雖然無法動态添加定時任務，但是可以在程式固定位置添加定時任務，如下：
from celery import Celery
from celery.schedules import crontab
app = Celery()
# 此處on_after_configure裝飾符意味着當Celery app配置完成之後調用該hook函數
@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)
這裡調用add_periodic_task用于添加一個定時任務，相當于在Celery config檔案中的beat_schedule設定項中添加了一項，如下：
app.conf.beat_schedule = {
'add-every-30-seconds': {
'task': 'tasks.add',
'schedule': 30.0,
'args': (16, 16)
},
}
在add_periodic_task中指定job function時需要用.s()來調用
使用APScheduler
筆者認為APScheduler是在實際項目最好用的一個工具庫。它不僅可以讓我們在程式中動态添加和删除我們的定時任務，還支援持久化，且其持久化方案支援很多形式，包括(Memory, MongoDB, SQLAlchemy, Redis, RethinkDB, ZooKeeper), 也可以非常好與一些Python framework內建(包括asyncio, gevent, Tornado, Twisted, Qt). 筆者所在的項目使用的是Flask架構，也有相應的插件可以供我們直接使用。
但是筆者沒有使用插件，而是直接将APScheduler內建于項目代碼中。
初始化scheduler
# 可以在初始化Flask的時候調用，并将傳回的scheduler賦給app
def init_scheduler():
# 這裡用于持久化的設定，代碼中示範使用MongoDB
# client用于設定你自己的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)
# 注意這裡一定要調用start啟動scheduler
scheduler.start()
return scheduler
添加定時任務
APScheduler将定時任務分為三種：
interval: 比如每隔5分鐘執行一次任務
cron: 比如每天早上5點執行一次任務
date: 比如在2018年5月5日執行一次任務
我們以添加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可以根據你自己的情況設定，其會被用于remove_job
current_app.scheduler.add_job(
test_job, 'cron', hour=hour, minute=minute,
args=[name], id=job_id)
删除定時任務
通過在add_job時使用的job_id可以删除對應的定時任務。實際上在我們添加任務的時候，APScheduler會把相應的任務資訊存儲于我們jobstore中設定的持久化存儲方案，這裡使用的是MongoDB，然後當删除的時候會将相應的任務從MongoDB中删除。
def remove_daily_job(job_id):
current_app.scheduler.remove_job(job_id)
總結：
APScheduler在實際使用過程中擁有最大的靈活性，可以滿足我們的大部分定時任務的相關需求；Celery比較重量級，通常如果項目中已有Celery在使用，而且不需要動态添加定時任務時可以考慮使用；schedule非常輕量級，使用簡單，但是不支援任務的持久化，也無法動态添加删除任務，是以主要用于簡單的小型應用。
References