Python中怎么利用APScheduler實現(xiàn)一個定時任務

今天就跟大家聊聊有關Python中怎么利用APScheduler實現(xiàn)一個定時任務，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結了以下內容，希望大家根據(jù)這篇文章可以有所收獲。

一、基本架構

1. 觸發(fā)器 triggers：設定觸發(fā)任務的條件

2. 描述一個任務何時被觸發(fā)，按日期或按時間間隔或按 cronjob 表達式三種方式觸發(fā)

3. 任務存儲器 job stores：存放任務，可以放內存(默認)或數(shù)據(jù)庫

4. 注：調度器之間不能共享任務存儲器

5. 執(zhí)行器 executors：用于執(zhí)行任務，可設定執(zhí)行模式

6. 將指定的作業(yè)提交到線程池或者進程池中運行，任務完成通知調度器觸發(fā)相應的事件。

7. 調度器 schedulers：將上方三個組件作為參數(shù)，創(chuàng)建調度器實例執(zhí)行。

協(xié)調三個組件的運行。

二、調度器組件(schedulers)

• BlockingScheduler 阻塞式調度器

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

from apscheduler.schedulers.blocking import BlockingScheduler
import time
scheduler = BlockingScheduler()
def job1():
 print "%s: 執(zhí)行任務" % time.asctime()
scheduler.add_job(job1, 'interval', seconds=3)
scheduler.start()

• BackgroundScheduler 后臺調度器

當前線程不會阻塞，調度器后臺執(zhí)行

from apscheduler.schedulers.background import BackgroundScheduler
import time
scheduler = BackgroundScheduler()
def job1():
 print "%s: 執(zhí)行任務" % time.asctime()
scheduler.add_job(job1, 'interval', seconds=3)
scheduler.start()
time.sleep(10)

注：10秒執(zhí)行完后，程序結束。

• AsyncIOScheduler AsyncIO調度器

適用于使用了asyncio的情況

from apscheduler.schedulers.asyncio import AsyncIOScheduler
import asyncio
...
...
try:
 asyncio.get_event_loop().run_forever()
except (KeyboardInterrupt, SystemExit):
 pass

• GeventScheduler Gevent調度器

使用了Gevent的情況

from apscheduler.schedulers.gevent import GeventScheduler
...
...
g = scheduler.start()
try:
 g.join()
except (KeyboardInterrupt, SystemExit):
 pass

• TornadoScheduler Tornado調度器

適用于構建Tornado應用

• TwistedScheduler Twisted調度器

適用于構建Twisted應用

• QtScheduler Qt調度器

適用于構建Qt應用

三、觸發(fā)器組件(trigger)

date ：只在某個時間點執(zhí)行一次，具體日期

run_date(datetime|str)

scheduler.add_job(my_job, 'date', run_date=datetime(2019, 7, 12, 15, 30, 5), args=[])
scheduler.add_job(my_job, 'date', run_date="2019-07-12", args=[])

timezone 指定時區(qū)

interval ：每隔一段時間允許一次，時間間隔

weeks=0 | days=0 | hours=0 | minutes=0 | seconds=0, start_date=None, end_date=None, timezone=None
scheduler.add_job(my_job, 'interval', hours=2)
scheduler.add_job(my_job, 'interval', hours=2, start_date='2017-9-8 21:30:00', end_date='2018-06-15 21:30:00)

cron ：任務的運行周期

(year=None, month=None, day=None, week=None, day_of_week=None, hour=None, minute=None, second=None, start_date=None, end_date=None, timezone=None)

除了week和 day_of_week，它們的默認值是 *

例如 day=1, minute=20 ，這就等于 year='*', month='*', day=1, week='*', day_of_week='*', hour='*', minute=20, second=0 ，工作將在每個月的第一天以每小時20分鐘的時間執(zhí)行

表達式類型

注：當設置的時間間隔小于，任務的執(zhí)行時間，線程會阻塞住，等待執(zhí)行完了才能執(zhí)行下一個任務，可以設置 max_instance 指定一個任務同一時刻有多少個實例在運行，默認為1

四、配置調度器

線程池執(zhí)行器默認為10，內存任務存儲器為 memoryjobstore ,如果想自己配置的話可以執(zhí)行以下操作

需求：

• 兩個任務儲存器分別搭配兩個執(zhí)行器；同時，還要修改任務的默認參數(shù)；最后還要改時區(qū)

• 名稱為“mongo”的 MongoDBJobStore

• 名稱為“default”的 SQLAlchemyJobStore

• 名稱為“ThreadPoolExecutor ”的 ThreadPoolExecutor ，最大線程20個

• 名稱“processpool”的 ProcessPoolExecutor ，最大進程5個

• UTC時間作為調度器的時區(qū)

• 默認為新任務關閉 合并模式 （）

• 設置新任務的默認最大實例數(shù)為3

方法一：

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
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(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)

方法二：

from apscheduler.schedulers.background import BackgroundScheduler
# The "apscheduler." prefix is hard coded
scheduler = BackgroundScheduler({
 'apscheduler.jobstores.mongo': {
   'type': 'mongodb'
 },
 'apscheduler.jobstores.default': {
  'type': 'sqlalchemy',
  'url': 'sqlite:///jobs.sqlite'
 },
 'apscheduler.executors.default': {
  'class': 'apscheduler.executors.pool:ThreadPoolExecutor',
  'max_workers': '20'
 },
 'apscheduler.executors.processpool': {
  'type': 'processpool',
  'max_workers': '5'
 },
 'apscheduler.job_defaults.coalesce': 'false',
 'apscheduler.job_defaults.max_instances': '3',
 'apscheduler.timezone': 'UTC',
})

方法三：

from pytz import utc
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ProcessPoolExecutor
jobstores = {
 'mongo': {'type': 'mongodb'},
 'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
executors = {
 'default': {'type': 'threadpool', 'max_workers': 20},
 'processpool': ProcessPoolExecutor(max_workers=5)
}
job_defaults = {
 'coalesce': False,
 'max_instances': 3
}
scheduler = BackgroundScheduler()
# ..這里可以添加任務
scheduler.configure(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)

五、啟動調度器

除了 BlockingScheduler 外，其他非阻塞的調度器都會立即返回，運行之后的代碼。

BlockingScheduler 需要將運行的代碼放在start()之前

1.添加任務

1.調用add_job()  #可以傳參max_instance,同一任務的運行實例個數(shù)

  當有任務中途中斷，后面恢復后，有N個任務沒有執(zhí)行 
    coalesce：true ，恢復的任務會執(zhí)行一次
    coalesce：false，恢復后的任務會執(zhí)行N次配合misfire_grace_time使用
    misfire_grace_time設置時間差值，由于某些原因沒有運行，再次提交時，大于設置的時間，實例不會運行。

2.裝飾器scheduled_job()

立即運行可以不設置trigger參數(shù)

2.移除任務

# 根據(jù)任務實例刪除
job = scheduler.add_job(myfunc, 'interval', minutes=2)
job.remove()
# 根據(jù)任務id刪除
scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
scheduler.remove_job('my_job_id')

3.暫停任務

job = scheduler.add_job(myfunc, 'interval', minutes=2)
# 根據(jù)任務實例
job.pause() #暫停
job.resume() #繼續(xù)
# 根據(jù)任務id暫停
scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
scheduler.pause_job('my_job_id')
scheduler.resume_job('my_job_id')

4.調度器操作

scheduler.start() #開啟
scheduler.shotdown(wait=True|False) #關閉 False 無論任務是否執(zhí)行，強制關閉

異常捕獲

# 可以添加apscheduler日志至DEBUG級別，這樣就能捕獲異常信息
import logging
logging.basicConfig()
logging.getLogger('apscheduler').setLevel(logging.DEBUG)

看完上述內容，你們對Python中怎么利用APScheduler實現(xiàn)一個定時任務有進一步的了解嗎？如果還想了解更多知識或者相關內容，請關注億速云行業(yè)資訊頻道，感謝大家的支持。