Python定時庫Apscheduler的簡單使用

Posted on by

        在Python中需要執行定時任務，可以使用Apscheduler。

        Apscheduler是基於Quartz的Python定時任務框架，功能上跟Quartz一致，使用上跟Quartz也幾乎一致。

        核心的四個部分：

        ①觸發器（trigger）、②作業存儲（job store）、③執行器（executor）、④調度器（scheduler）

        安裝依賴：

pip install apscheduler

        間隔時間調度：

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
 
sched = BlockingScheduler()

def test_job():
    print(f'{datetime.now():%H:%M:%S} Test job')

 
if __name__ == '__main__':
 
    sched.add_job(test_job, 'interval', id='test', seconds=5)
    sched.start()

        也可以使用裝飾器：

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
 
 
sched = BlockingScheduler()
 
 
def test_job():
    print(f'{datetime.now():%H:%M:%S} Test job')
 
 
@sched.scheduled_job('interval', seconds=5)
def test_decorator_job():
    print(f'{datetime.now():%H:%M:%S} Test decorator job')
 
 
if __name__ == '__main__':
 
    sched.add_job(test_job, 'interval', id='test', seconds=5)
    sched.start()

        運行結果：

        BlockingScheduler()是調度器中的一種調度器

        sched.add_job()是添加作業

        sched.start()是開始任務

         定時調度：

        定時調度使用cron表達式進行，這裡也帶參數執行：

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
 
scheduler = BlockingScheduler()
 
 
def test_args(x):
    print (f'{datetime.now():%H:%M:%S} Test cron job', x)
 
 
if __name__ == '__main__':
    scheduler.add_job(test_args, 'cron', args=('cron test',), second='*/5')
    scheduler.start()

        時間參數設置如下：

year (int|str) – 年，4位數字 
month (int|str) – 月 (范圍1-12) 
day (int|str) – 日 (范圍1-31) 
week (int|str) – 周 (范圍1-53) 
day_of_week (int|str) – 周內第幾天或者星期幾 (范圍0-6 或者 mon,tue,wed,thu,fri,sat,sun) 
hour (int|str) – 時 (范圍0-23) 
minute (int|str) – 分 (范圍0-59) 
second (int|str) – 秒 (范圍0-59) 
start_date (datetime|str) – 最早開始日期(包含) 
end_date (datetime|str) – 最晚結束時間(包含) 
timezone (datetime.tzinfo|str) – 指定時區

        接下來說一下其中的調度器：

BlockingScheduler：適用於調度程序是進程中唯一運行的進程，調用start函數會阻塞當前線程，不能立即返回。
BackgroundScheduler：適用於調度程序在應用程序的後臺運行，調用start後主線程不會阻塞。
AsyncIOScheduler：適用於使用瞭asyncio模塊的應用程序。
GeventScheduler：適用於使用gevent模塊的應用程序。
TwistedScheduler：適用於構建Twisted的應用程序。
QtScheduler：適用於構建Qt的應用程序。

        其中使用得比較多的是前三種調度器。

        刪除任務：

scheduler.remove_job(‘task_id’)

        停止任務：

scheduler.pause_job(‘task_id’)

        恢復任務：

scheduler.resume_job(‘task_id’)

  立即執行任務next_run_time：

scheduler.add_job(
            test_job,
            'interval',
            minutes=5,
            next_run_time=datetime.datetime.now()
        )

到此這篇關於Python定時庫Apscheduler的簡單使用的文章就介紹到這瞭,更多相關Python定時庫Apscheduler使用內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

推薦閱讀：