Python怎么使用APScheduler調度任務

這篇文章主要介紹“Python怎么使用APScheduler調度任務”，在日常操作中，相信很多人在Python怎么使用APScheduler調度任務問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”Python怎么使用APScheduler調度任務”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！

在丹陽等地區(qū)，都構建了全面的區(qū)域性戰(zhàn)略布局，加強發(fā)展的系統(tǒng)性、市場前瞻性、產品創(chuàng)新能力，以專注、極致的服務理念，為客戶提供成都網站建設、網站制作 網站設計制作按需求定制設計,公司網站建設,企業(yè)網站建設,品牌網站建設,網絡營銷推廣,外貿網站建設,丹陽網站建設費用合理。

 任務調度應用場景

所謂的任務調度是指安排任務的執(zhí)行計劃，即何時執(zhí)行，怎么執(zhí)行等。在現(xiàn)實項目中經常出現(xiàn)它們的身影；特別是數(shù)據(jù)類項目，比如實時統(tǒng)計每5分鐘網站的訪問量，就需要每5分鐘定時從日志數(shù)據(jù)分析訪問量。

總結下任務調度應用場景：

  •  離線作業(yè)調度：按時間粒度執(zhí)行某項任務

  •  共享緩存更新：定時刷新緩存，如redis緩存；不同進程間的共享數(shù)據(jù)

任務調度工具

  •  linux的crontab， 支持按照分鐘/小時/天/月/周粒度，執(zhí)行任務

  •  java的Quartz

  •  windows的任務計劃

本文介紹的是python中的任務調度庫，APScheduler（advance python scheduler）。如果你了解Quartz的話，可以看出APScheduler是Quartz的python實現(xiàn)；APScheduler提供了基于時間，固定時間點和crontab方式的任務調用方案， 可以當作一個跨平臺的調度工具來使用。

APScheduler

組件介紹

APScheduler由5個部分組成：觸發(fā)器、調度器、任務存儲器、執(zhí)行器和任務事件。

  •  任務job：任務id和任務執(zhí)行func

  •  觸發(fā)器triggers：確定任務何時開始執(zhí)行

  •  任務存儲器job stores: 保存任務的狀態(tài)

  •  執(zhí)行器executors：確定任務怎么執(zhí)行

  •  任務事件event：監(jiān)控任務執(zhí)行異常情況

  •  調度器schedulers：串聯(lián)任務的整個生命周期，添加編輯任務到任務存儲器,在任務的執(zhí)行時間到來時，把任務交給執(zhí)行器執(zhí)行返回結果；同時發(fā)出事件監(jiān)聽，監(jiān)控任務事件 。

安裝

pip install apscheduler

簡單例子

from apscheduler.schedulers.background import BackgroundScheduler  from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor  from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore  from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR  import logging  import datetime  # 任務執(zhí)行函數(shù)  def job_func(job_id):      print('job %s is runed at %s' % (job_id, datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')))  # 事件監(jiān)聽  def job_exception_listener(event):      if event.exception:          # todo：異常處理, 告警等          print('The job crashed :(')      else:          print('The job worked :)')  # 日志  logging.basicConfig()  logging.getLogger('apscheduler').setLevel(logging.DEBUG)  # 定義一個后臺任務非阻塞調度器  scheduler = BackgroundScheduler()  # 添加一個任務到內存中   # 觸發(fā)器：trigger='interval' seconds=10 每10s觸發(fā)執(zhí)行一次  # 執(zhí)行器：executor='default' 線程執(zhí)行  # 任務存儲器：jobstore='default' 默認內存存儲  # 最大并發(fā)數(shù)：max_instances  scheduler.add_job(job_func, trigger='interval', args=[1], id='1', name='a test job', max_instances=10, jobstore='default', executor='default', seconds=10)  # 設置任務監(jiān)聽  scheduler.add_listener(job_exception_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)  # 啟動調度器  scheduler.start()

運行情況：

job 1 is runed at 2020-03-21 20:00:38  The job worked :)  job 1 is runed at 2020-03-21 20:00:48  The job worked :)  job 1 is runed at 2020-03-21 20:00:58  The job worked :)

觸發(fā)器

觸發(fā)器決定何時執(zhí)行任務，APScheduler支持的觸發(fā)器有3種

  •  trigger='interval'：按固定時間周期執(zhí)行，支持weeks，days，hours，minutes， seconds， 還可指定時間范圍   

sched.add_job(job_function, 'interval', hours=2, start_date='2010-10-10 09:30:00', end_date='2014-06-15 11:00:00')

  •  trigger='date': 固定時間，執(zhí)行一次   

sched.add_job(my_job, 'date', run_date=datetime(2009, 11, 6, 16, 30, 5), args=['text'])

  •   trigger='cron': 支持crontab方式，執(zhí)行任務

  •   參數(shù)：分鐘/小時/天/月/周粒度，也可指定時間范圍     

year (int|str) – 4-digit year        month (int|str) – month (1-12)        day (int|str) – day of the (1-31)        week (int|str) – ISO week (1-53)        day_of_week (int|str) – number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun)        hour (int|str) – hour (0-23)        minute (int|str) – minute (0-59)        second (int|str) – second (0-59)        start_date (datetime|str) – earliest possible date/time to trigger on (inclusive)        end_date (datetime|str) – latest possible date/time to trigger on (inclusive)

  •   例子           

# 星期一到星期五，5點30執(zhí)行任務job_function，直到2014-05-30 00:00:00             sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2014-05-30')             # 按照crontab格式執(zhí)行， 格式為：分鐘 小時 天 月 周，*表示所有             # 5月到8月的1號到15號，0點0分執(zhí)行任務job_function             sched.add_job(job_function, CronTrigger.from_crontab('0 0 1-15 may-aug *'))

執(zhí)行器

執(zhí)行器決定如何執(zhí)行任務；APScheduler支持4種不同執(zhí)行器，常用的有pool(線程/進程)和gevent(io多路復用，支持高并發(fā))，默認為pool中線程池， 不同的執(zhí)行器可以在調度器的配置中進行配置（見調度器）

  •  apscheduler.executors.asyncio：同步io，阻塞

  •  apscheduler.executors.gevent：io多路復用，非阻塞

  •  apscheduler.executors.pool: 線程ThreadPoolExecutor和進程ProcessPoolExecutor

  •  apscheduler.executors.twisted：基于事件驅動

任務存儲器

任務存儲器決定任務的保存方式， 默認存儲在內存中（MemoryJobStore），重啟后就沒有了。APScheduler支持的任務存儲器有：

  •  apscheduler.jobstores.memory：內存

  •  apscheduler.jobstores.MongoDB：存儲在mongodb

  •  apscheduler.jobstores.redis：存儲在redis

  •  apscheduler.jobstores.rethinkdb：存儲在rethinkdb

  •  apscheduler.jobstores.sqlalchemy：支持sqlalchemy的數(shù)據(jù)庫如MySQL，sqlite等

  •  apscheduler.jobstores.zookeeper：zookeeper

不同的任務存儲器可以在調度器的配置中進行配置（見調度器）

調度器

APScheduler支持的調度器方式如下，比較常用的為BlockingScheduler和BackgroundScheduler

  •  BlockingScheduler：適用于調度程序是進程中唯一運行的進程，調用start函數(shù)會阻塞當前線程，不能立即返回。

  •  BackgroundScheduler：適用于調度程序在應用程序的后臺運行，調用start后主線程不會阻塞。

  •  AsyncIOScheduler：適用于使用了asyncio模塊的應用程序。

  •  GeventScheduler：適用于使用gevent模塊的應用程序。

  •  TwistedScheduler：適用于構建Twisted的應用程序。

  •  QtScheduler：適用于構建Qt的應用程序。

從前面的例子，我們可以看到，調度器可以操作任務（并為任務指定觸發(fā)器、任務存儲器和執(zhí)行器）和監(jiān)控任務。

scheduler.add_job(job_func, trigger='interval', args=[1], id='1', name='a test job', max_instances=10, jobstore='default', executor='default', seconds=10)

我們來詳細看下各個部分

  •  調度器配置：在add_job我們看到jobstore和executor都是default，APScheduler在定義調度器時可以指定不同的任務存儲和執(zhí)行器，以及初始的參數(shù)   

from pytz import utc     from apscheduler.schedulers.background import BackgroundScheduler     from apscheduler.jobstores.mongodb import MongoDBJobStore     from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore     from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor     # 通過dict方式執(zhí)行不同的jobstores、executors和默認的參數(shù)     jobstores = {         'mongo': MongoDBJobStore(),         'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')     }     executors = {         'default': ThreadPoolExecutor(20),         'processpool': ProcessPoolExecutor(5)     }     job_defaults = {         'coalesce': False,         'max_instances': 3     }     # 定義調度器     scheduler = BackgroundScheduler(jobstoresjobstores=jobstores, executorsexecutors=executors, job_defaultsjob_defaults=job_defaults, timezone=utc)     def job_func(job_id):         print('job %s is runed at %s' % (job_id, datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')))     # 添加任務     scheduler.add_job(job_func, trigger='interval', args=[1], id='1', name='a test job', jobstore='default', executor='processpool', seconds=10)     # 啟動調度器     scheduler.start()

  •  操作任務：調度器可以增加，刪除，暫停，恢復和修改任務。需要注意的是這里的操作只是對未執(zhí)行的任務起作用，已經執(zhí)行和正在執(zhí)行的任務不受這些操作的影響。

  •   add_job     

scheduler.add_job(job_func, trigger='interval', args=[1], id='1', name='a test job', max_instances=10, jobstore='default', executor='default', seconds=10)

  •   remove_job: 通過任務唯一的id，刪除的時候對應的任務存儲器里記錄也會刪除 

scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')   scheduler.remove_job('my_job_id')

  •   Pausing and resuming jobs：暫停和重啟任務       

scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')          scheduler.pause_job('my_job_id')          scheduler.resume_job('my_job_id')

  •   Modifying jobs：修改任務的配置       

job = scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id', max_instances=10)         # 修改任務的屬性         job.modify(max_instances=6, name='Alternate name')         # 修改任務的觸發(fā)器         scheduler.reschedule_job('my_job_id', trigger='cron', minute='*/5')

  •  監(jiān)控任務事件類型，比較常用的類型有：

    •   EVENT_JOB_ERROR: 表示任務在執(zhí)行過程的出現(xiàn)異常觸發(fā)

    •   EVENT_JOB_EXECUTED：任務執(zhí)行成功時

    •   EVENT_JOB_MAX_INSTANCES：調度器上執(zhí)行的任務超過配置的參數(shù)時       

scheduler.add_listener(job_exception_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)

到此，關于“Python怎么使用APScheduler調度任務”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續(xù)學習更多相關知識，請繼續(xù)關注創(chuàng)新互聯(lián)網站，小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>

本文題目：Python怎么使用APScheduler調度任務
地址分享：http://muchs.cn/article22/jiocjc.html

成都網站建設公司_創(chuàng)新互聯(lián)，為您提供企業(yè)網站制作、域名注冊、標簽優(yōu)化、自適應網站、全網營銷推廣、商城網站

廣告

聲明：本網站發(fā)布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經允許不得轉載，或轉載時需注明來源： 創(chuàng)新互聯(lián)

外貿網站建設