Python定時庫Apscheduler的簡單使用
在Python中需要執(zhí)行定時任務,可以使用Apscheduler。
Apscheduler是基于Quartz的Python定時任務框架,功能上跟Quartz一致,使用上跟Quartz也幾乎一致。
核心的四個部分:
①觸發(fā)器(trigger)、②作業(yè)存儲(job store)、③執(zhí)行器(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()是添加作業(yè)
sched.start()是開始任務
定時調度:
定時調度使用cron表達式進行,這里也帶參數執(zhí)行:
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) – 指定時區(qū)
接下來說一下其中的調度器:
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')
立即執(zhí)行任務next_run_time:
scheduler.add_job( test_job, 'interval', minutes=5, next_run_time=datetime.datetime.now() )
到此這篇關于Python定時庫Apscheduler的簡單使用的文章就介紹到這了,更多相關Python定時庫Apscheduler使用內容請搜索本站以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持本站!
版權聲明:本站文章來源標注為YINGSOO的內容版權均為本站所有,歡迎引用、轉載,請保持原文完整并注明來源及原文鏈接。禁止復制或仿造本網站,禁止在非www.sddonglingsh.com所屬的服務器上建立鏡像,否則將依法追究法律責任。本站部分內容來源于網友推薦、互聯(lián)網收集整理而來,僅供學習參考,不代表本站立場,如有內容涉嫌侵權,請聯(lián)系alex-e#qq.com處理。