今天準(zhǔn)備實(shí)現(xiàn)一個(gè)功能需要用到定時(shí)執(zhí)行任務(wù)，所以就看到了Python的一個(gè)定時(shí)任務(wù)框架APScheduler，試了一下感覺還不錯(cuò)。

1.APScheduler簡(jiǎn)介：

APScheduler是Python的一個(gè)定時(shí)任務(wù)框架，可以很方便的滿足用戶定時(shí)執(zhí)行或者周期執(zhí)行任務(wù)的需求，它提供了基于日期date、固定時(shí)間間隔interval 、以及類似于Linux上的定時(shí)任務(wù)crontab類型的定時(shí)任務(wù)。并且該框架不僅可以添加、刪除定時(shí)任務(wù)，還可以將任務(wù)存儲(chǔ)到數(shù)據(jù)庫中，實(shí)現(xiàn)任務(wù)的持久化，所以使用起來非常方便。

2.APScheduler安裝：

APScheduler的安裝相對(duì)來說也非常簡(jiǎn)單，可以直接利用pip安裝，如果沒有pip可以下載源碼，利用源碼安裝。

1).利用pip安裝：(推薦)#?pip?install?apscheduler#?python?setup.py?install

3.基本概念

APScheduler有四種組件及相關(guān)說明：

1) triggers(觸發(fā)器)：觸發(fā)器包含調(diào)度邏輯，每一個(gè)作業(yè)有它自己的觸發(fā)器，用于決定接下來哪一個(gè)作業(yè)會(huì)運(yùn)行，除了他們自己初始化配置外，觸發(fā)器完全是無狀態(tài)的。

2)job stores(作業(yè)存儲(chǔ))：用來存儲(chǔ)被調(diào)度的作業(yè)，默認(rèn)的作業(yè)存儲(chǔ)器是簡(jiǎn)單地把作業(yè)任務(wù)保存在內(nèi)存中，其它作業(yè)存儲(chǔ)器可以將任務(wù)作業(yè)保存到各種數(shù)據(jù)庫中，支持MongoDB、Redis、SQLAlchemy存儲(chǔ)方式。當(dāng)對(duì)作業(yè)任務(wù)進(jìn)行持久化存儲(chǔ)的時(shí)候，作業(yè)的數(shù)據(jù)將被序列化，重新讀取作業(yè)時(shí)在反序列化。

3) executors(執(zhí)行器)：執(zhí)行器用來執(zhí)行定時(shí)任務(wù)，只是將需要執(zhí)行的任務(wù)放在新的線程或者線程池中運(yùn)行。當(dāng)作業(yè)任務(wù)完成時(shí)，執(zhí)行器將會(huì)通知調(diào)度器。對(duì)于執(zhí)行器，默認(rèn)情況下選擇ThreadPoolExecutor就可以了，但是如果涉及到一下特殊任務(wù)如比較消耗CPU的任務(wù)則可以選擇ProcessPoolExecutor，當(dāng)然根據(jù)根據(jù)實(shí)際需求可以同時(shí)使用兩種執(zhí)行器。

4) schedulers(調(diào)度器)：調(diào)度器是將其它部分聯(lián)系在一起，一般在應(yīng)用程序中只有一個(gè)調(diào)度器，應(yīng)用開發(fā)者不會(huì)直接操作觸發(fā)器、任務(wù)存儲(chǔ)以及執(zhí)行器，相反調(diào)度器提供了處理的接口。通過調(diào)度器完成任務(wù)的存儲(chǔ)以及執(zhí)行器的配置操作，如可以添加。修改、移除任務(wù)作業(yè)。

APScheduler提供了多種調(diào)度器，可以根據(jù)具體需求來選擇合適的調(diào)度器，常用的調(diào)度器有：

BlockingScheduler：適合于只在進(jìn)程中運(yùn)行單個(gè)任務(wù)的情況，通常在調(diào)度器是你唯一要運(yùn)行的東西時(shí)使用。

BackgroundScheduler: 適合于要求任何在程序后臺(tái)運(yùn)行的情況，當(dāng)希望調(diào)度器在應(yīng)用后臺(tái)執(zhí)行時(shí)使用。

AsyncIOScheduler：適合于使用asyncio框架的情況

GeventScheduler: 適合于使用gevent框架的情況

TornadoScheduler: 適合于使用Tornado框架的應(yīng)用

TwistedScheduler: 適合使用Twisted框架的應(yīng)用

QtScheduler: 適合使用QT的情況

4.配置調(diào)度器

APScheduler提供了許多不同的方式來配置調(diào)度器，你可以使用一個(gè)配置字典或者作為參數(shù)關(guān)鍵字的方式傳入。你也可以先創(chuàng)建調(diào)度器，再配置和添加作業(yè)，這樣你可以在不同的環(huán)境中得到更大的靈活性。

1)下面一個(gè)簡(jiǎn)單的示例：import?time

from?apscheduler.schedulers.blocking?import?BlockingScheduler

def?test_job():

print?time.strftime('%Y-%m-%d?%H:%M:%S',?time.localtime(time.time()))

scheduler?=?BlockingScheduler()

'''

#該示例代碼生成了一個(gè)BlockingScheduler調(diào)度器，使用了默認(rèn)的默認(rèn)的任務(wù)存儲(chǔ)MemoryJobStore，以及默認(rèn)的執(zhí)行器ThreadPoolExecutor，并且最大線程數(shù)為10。

'''

scheduler.add_job(test_job,?'interval',?seconds=5,?id='test_job')

'''

#該示例中的定時(shí)任務(wù)采用固定時(shí)間間隔(interval)的方式，每隔5秒鐘執(zhí)行一次。

#并且還為該任務(wù)設(shè)置了一個(gè)任務(wù)id

'''

scheduler.start()

2)如果想執(zhí)行一些復(fù)雜任務(wù)，如上邊所說的同時(shí)使用兩種執(zhí)行器，或者使用多種任務(wù)存儲(chǔ)方式，并且需要根據(jù)具體情況對(duì)任務(wù)的一些默認(rèn)參數(shù)進(jìn)行調(diào)整。可以參考下面的方式。(http://apscheduler.readthedocs.io/en/latest/userguide.html)

第一種方式：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()

#?..?do?something?else?here,?maybe?add?jobs?etc.

scheduler.configure(jobstores=jobstores,?executors=executors,?job_defaults=job_defaults,?timezone=utc)

5.對(duì)任務(wù)作業(yè)的基本操作：

1).添加作業(yè)有兩種方式：第一種可以直接調(diào)用add_job()，第二種使用scheduled_job()修飾器。

而add_job()是使用最多的，它可以返回一個(gè)apscheduler.job.Job實(shí)例，因而可以對(duì)它進(jìn)行修改或者刪除，而使用修飾器添加的任務(wù)添加之后就不能進(jìn)行修改。

例如使用add_job()添加作業(yè)：#!/usr/bin/env?python

#-*-?coding:UTF-8

import?time

import?datetime

from?apscheduler.schedulers.blocking?import?BlockingScheduler

def?job1(f):

print?time.strftime('%Y-%m-%d?%H:%M:%S',?time.localtime(time.time())),?f

def?job2(arg1,?args2,?f):

print?f,?args1,?args2

def?job3(**args):

print?args

'''

APScheduler支持以下三種定時(shí)任務(wù)：

cron:?crontab類型任務(wù)

interval:?固定時(shí)間間隔任務(wù)

date:?基于日期時(shí)間的一次性任務(wù)

'''

scheduler?=?BlockingScheduler()

#循環(huán)任務(wù)示例

scheduler.add_job(job1,?'interval',?seconds=5,?args=('循環(huán)',),?id='test_job1')

#定時(shí)任務(wù)示例

scheduler.add_job(job1,?'cron',?second='*/5',?args=('定時(shí)',),?id='test_job2')

#一次性任務(wù)示例

scheduler.add_job(job1,?next_run_time=(datetime.datetime.now()?+?datetime.timedelta(seconds=10)),?args=('一次',),?id='test_job3')

'''

傳遞參數(shù)的方式有元組(tuple)、列表(list)、字典(dict)

注意：不過需要注意采用元組傳遞參數(shù)時(shí)后邊需要多加一個(gè)逗號(hào)

'''

#基于list

scheduler.add_job(job2,?'interval',?seconds=5,?args=['a','b','list'],?id='test_job4')

#基于tuple

scheduler.add_job(job2,?'interval',?seconds=5,?args=('a','b','tuple',),?id='test_job5')

#基于dict

scheduler.add_job(job3,?'interval',?seconds=5,?kwargs={'f':'dict',?'a':1,'b':2},?id='test_job7')

print?scheduler.get_jobs()

scheduler.start()

#帶有參數(shù)的示例

scheduler.add_job(job2,?'interval',?seconds=5,?args=['a','b'],?id='test_job4')

scheduler.add_job(job2,?'interval',?seconds=5,?args=('a','b',),?id='test_job5')

scheduler.add_job(job3,?'interval',?seconds=5,?kwargs={'a':1,'b':2},?id='test_job6')

print?scheduler.get_jobs()

scheduler.start()

或者使用scheduled_job()修飾器來添加作業(yè)：@sched.scheduled_job('cron',?second='*/5'?，id='my_job_id',)

def?test_task():

print("Hello?world!")

2).獲得任務(wù)列表：

可以通過get_jobs方法來獲取當(dāng)前的任務(wù)列表，也可以通過get_job()來根據(jù)job_id來獲得某個(gè)任務(wù)的信息。并且apscheduler還提供了一個(gè)print_jobs()方法來打印格式化的任務(wù)列表。

例如：scheduler.add_job(my_job,?'interval',?seconds=5,?id='my_job_id'?name='test_job')

print?scheduler.get_job('my_job_id')

print?scheduler.get_jobs()

3).修改任務(wù)：

修改任務(wù)任務(wù)的屬性可以使用apscheduler.job.Job.modify()或者modify_job()方法，可以修改除了id的其它任何屬性。

例如：job?=?scheduler.add_job(my_job,?'interval',?seconds=5,?id='my_job'?name='test_job')

job.modify(max_instances=5,?name='my_job')

4).刪除任務(wù):

刪除調(diào)度器中的任務(wù)有可以用remove_job()根據(jù)job ID來刪除指定任務(wù)或者使用remove()，如果使用remove()需要事先保存在添加任務(wù)時(shí)返回的實(shí)例對(duì)象，任務(wù)刪除后就不會(huì)在執(zhí)行。

注意：通過scheduled_job()添加的任務(wù)只能使用remove_job()進(jìn)行刪除。

例如：job?=?scheduler.add_job(my_job,?'interval',?seconds=5,?id='my_job_id'?name='test_job')

job.remove()

或者scheduler.add_job(my_job,?'interval',?seconds=5,?id='my_job_id'?name='test_job')

scheduler.remove_job('my_job')

5).暫停與恢復(fù)任務(wù)：

暫停與恢復(fù)任務(wù)可以直接操作任務(wù)實(shí)例或者調(diào)度器來實(shí)現(xiàn)。當(dāng)任務(wù)暫停時(shí)，它的運(yùn)行時(shí)間會(huì)被重置，暫停期間不會(huì)計(jì)算時(shí)間。

暫停任務(wù)：apscheduler.job.Job.pause()

apscheduler.schedulers.base.BaseScheduler.pause_job()

恢復(fù)任務(wù)apscheduler.job.Job.resume()

apscheduler.schedulers.BaseScheduler.resume_job()

6).啟動(dòng)調(diào)度器

可以使用start()方法啟動(dòng)調(diào)度器，BlockingScheduler需要在初始化之后才能執(zhí)行start(),對(duì)于其他的Scheduler，調(diào)用start()方法都會(huì)直接返回，然后可以繼續(xù)執(zhí)行后面的初始化操作。

例如:from?apscheduler.schedulers.blocking?import?BlockingScheduler

def?my_job():

print?"Hello?world!"

scheduler?=?BlockingScheduler()

scheduler.add_job(my_job,?'interval',?seconds=5)

scheduler.start()

7).關(guān)閉調(diào)度器：

使用下邊方法關(guān)閉調(diào)度器：scheduler.shutdown()

默認(rèn)情況下調(diào)度器會(huì)關(guān)閉它的任務(wù)存儲(chǔ)和執(zhí)行器，并等待所有正在執(zhí)行的任務(wù)完成，如果不想等待，可以進(jìn)行如下操作：scheduler.shutdown(wait=False)

注意：

當(dāng)出現(xiàn)No handlers could be found for logger “apscheduler.scheduler”次錯(cuò)誤信息時(shí)，說明沒有

logging模塊的logger存在，所以需要添加上，對(duì)應(yīng)新增內(nèi)容如下所示(僅供參考)：import?logging

logging.basicConfig(level=logging.DEBUG,

format='%(asctime)s?%(filename)s[line:%(lineno)d]?%(levelname)s?%(message)s',

datafmt='%a,?%d?%b?%Y?%H:%M:%S',

filename='/var/log/aaa.txt',

filemode='a')

總結(jié)

以上是生活随笔為你收集整理的python apscheduler执行_Python下定时任务框架APScheduler的使用的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。