在Python開發過程中我們經常需要執行定時任務，而此類任務我們通常有如下選項：

• 自己造輪子
• 使用schedule庫
• 使用Celery定時任務
• 使用APScheduler

自己造輪子實現，最大的優勢就是靈活性，調試方便，對于某些特定系統也許也是一種選擇，不過對于大多數應用來說，我們應當盡可能地使用開源的成熟的方案。下面對后三種方案分別討論：

使用schedule庫

schedule庫是一個輕量級的定時任務方案，優勢是使用簡單，也不需要做什么配置；缺點是無法動態添加任務，也無法將任務持久化。

安裝

?

pip install schedule

使用

?

import schedule import timedef 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 crontabapp = 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 secondssender.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啟動schedulerscheduler.start()return scheduler

添加定時任務

APScheduler將定時任務分為三種：

• interval: 比如每隔5分鐘執行一次任務
• cron: 比如每天早上5點執行一次任務
• date: 比如在2018年5月5日執行一次任務

我們以添加cron job為例：

?

def test_job(name):print "hello, %s" % namedef 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_jobcurrent_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

• Schedule
• Celery
• APScheduler

作者：geekpy
鏈接：https://www.jianshu.com/p/94b273f6ed77
來源：簡書
著作權歸作者所有。商業轉載請聯系作者獲得授權，非商業轉載請注明出處。

總結

以上是生活随笔為你收集整理的Python定时任务-schedule vs. Celery vs. APScheduler的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。