Python定時任務-schedule vs. Celery vs. APScheduler

在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

• Schedule
• Celery
• APScheduler

作者：geekpy

連結：https://www.jianshu.com/p/94b273f6ed77

來源：簡書

著作權歸作者所有。商業轉載請聯系作者獲得授權，非商業轉載請注明出處。