在Python開(kāi)發(fā)過(guò)程中我們經(jīng)常需要執(zhí)行定時(shí)任務(wù)，而此類(lèi)任務(wù)我們通常有如下選項(xiàng)：

• 自己造輪子
• 使用schedule庫(kù)
• 使用Celery定時(shí)任務(wù)
• 使用APScheduler

自己造輪子實(shí)現(xiàn)，最大的優(yōu)勢(shì)就是靈活性，調(diào)試方便，對(duì)于某些特定系統(tǒng)也許也是一種選擇，不過(guò)對(duì)于大多數(shù)應(yīng)用來(lái)說(shuō)，我們應(yīng)當(dāng)盡可能地使用開(kāi)源的成熟的方案。下面對(duì)后三種方案分別討論：

使用schedule庫(kù)

schedule庫(kù)是一個(gè)輕量級(jí)的定時(shí)任務(wù)方案，優(yōu)勢(shì)是使用簡(jiǎn)單，也不需要做什么配置；缺點(diǎn)是無(wú)法動(dòng)態(tài)添加任務(wù)，也無(wú)法將任務(wù)持久化。

安裝

?

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領(lǐng)域可謂大名鼎鼎，我們通常將Celery作為一個(gè)任務(wù)隊(duì)列來(lái)使用，不過(guò)Celery也同時(shí)提供了定時(shí)任務(wù)功能。通常，當(dāng)我們的解決方案中已經(jīng)在使用Celery的時(shí)候可以考慮同時(shí)使用其定時(shí)任務(wù)功能，但是Celery無(wú)法在Flask這樣的系統(tǒng)中動(dòng)態(tài)添加定時(shí)任務(wù)（在Django中有相應(yīng)的插件可以實(shí)現(xiàn)動(dòng)態(tài)添加任務(wù)），而且如果對(duì)于不使用Celery的項(xiàng)目，單獨(dú)為定時(shí)任務(wù)搭建Celery顯得過(guò)于重量級(jí)了。(搭建Celery比較麻煩，還需要配置諸如RabbitMQ之類(lèi)消息分發(fā)程序)。

Celery安裝在此不再贅述，大家可以參考官網(wǎng)的資料

使用

Celery雖然無(wú)法動(dòng)態(tài)添加定時(shí)任務(wù)，但是可以在程序固定位置添加定時(shí)任務(wù)，如下：

?

from celery import Celery from celery.schedules import crontabapp = Celery()# 此處on_after_configure裝飾符意味著當(dāng)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 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)

• 這里調(diào)用add_periodic_task用于添加一個(gè)定時(shí)任務(wù)，相當(dāng)于在Celery config文件中的beat_schedule設(shè)置項(xiàng)中添加了一項(xiàng)，如下： app.conf.beat_schedule = {'add-every-30-seconds': {'task': 'tasks.add','schedule': 30.0,'args': (16, 16)}, }
• 在add_periodic_task中指定job function時(shí)需要用.s()來(lái)調(diào)用

使用APScheduler

筆者認(rèn)為APScheduler是在實(shí)際項(xiàng)目最好用的一個(gè)工具庫(kù)。它不僅可以讓我們?cè)诔绦蛑袆?dòng)態(tài)添加和刪除我們的定時(shí)任務(wù)，還支持持久化，且其持久化方案支持很多形式，包括(Memory, MongoDB, SQLAlchemy, Redis, RethinkDB, ZooKeeper), 也可以非常好與一些Python framework集成(包括asyncio, gevent, Tornado, Twisted, Qt). 筆者所在的項(xiàng)目使用的是Flask框架，也有相應(yīng)的插件可以供我們直接使用。

但是筆者沒(méi)有使用插件，而是直接將APScheduler集成于項(xiàng)目代碼中。

初始化scheduler

?

# 可以在初始化Flask的時(shí)候調(diào)用，并將返回的scheduler賦給app def init_scheduler():# 這里用于持久化的設(shè)置，代碼中演示使用MongoDB# client用于設(shè)置你自己的MongoDB的handler, 即MongoClient對(duì)象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啟動(dòng)schedulerscheduler.start()return scheduler

添加定時(shí)任務(wù)

APScheduler將定時(shí)任務(wù)分為三種：

• interval: 比如每隔5分鐘執(zhí)行一次任務(wù)
• cron: 比如每天早上5點(diǎn)執(zhí)行一次任務(wù)
• date: 比如在2018年5月5日?qǐng)?zhí)行一次任務(wù)

我們以添加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可以根據(jù)你自己的情況設(shè)定，其會(huì)被用于remove_jobcurrent_app.scheduler.add_job(test_job, 'cron', hour=hour, minute=minute,args=[name], id=job_id)

刪除定時(shí)任務(wù)

通過(guò)在add_job時(shí)使用的job_id可以刪除對(duì)應(yīng)的定時(shí)任務(wù)。實(shí)際上在我們添加任務(wù)的時(shí)候，APScheduler會(huì)把相應(yīng)的任務(wù)信息存儲(chǔ)于我們jobstore中設(shè)置的持久化存儲(chǔ)方案，這里使用的是MongoDB，然后當(dāng)刪除的時(shí)候會(huì)將相應(yīng)的任務(wù)從MongoDB中刪除。

?

def remove_daily_job(job_id):current_app.scheduler.remove_job(job_id)

總結(jié)：

APScheduler在實(shí)際使用過(guò)程中擁有最大的靈活性，可以滿足我們的大部分定時(shí)任務(wù)的相關(guān)需求；Celery比較重量級(jí)，通常如果項(xiàng)目中已有Celery在使用，而且不需要?jiǎng)討B(tài)添加定時(shí)任務(wù)時(shí)可以考慮使用；schedule非常輕量級(jí)，使用簡(jiǎn)單，但是不支持任務(wù)的持久化，也無(wú)法動(dòng)態(tài)添加刪除任務(wù)，所以主要用于簡(jiǎn)單的小型應(yīng)用。

References

• Schedule
• Celery
• APScheduler

作者：geekpy
鏈接：https://www.jianshu.com/p/94b273f6ed77
來(lái)源：簡(jiǎn)書(shū)
著作權(quán)歸作者所有。商業(yè)轉(zhuǎn)載請(qǐng)聯(lián)系作者獲得授權(quán)，非商業(yè)轉(zhuǎn)載請(qǐng)注明出處。

總結(jié)

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

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。