前言

APScheduler是基于Quartz的一個(gè)Python定時(shí)任務(wù)框架。提供了基于日期、固定時(shí)間間隔以及crontab類(lèi)型的任務(wù)，并且可以持久化任務(wù)。

一、安裝APScheduler

pip install apscheduler

二、基本概念

APScheduler有四大組件：

1、觸發(fā)器 triggers ：

觸發(fā)器包含調(diào)度邏輯。每個(gè)作業(yè)都有自己的觸發(fā)器，用于確定下一個(gè)任務(wù)何時(shí)運(yùn)行。除了初始配置之外，觸發(fā)器是完全無(wú)狀態(tài)的。

有三種內(nèi)建的trigger:

(1)date: 特定的時(shí)間點(diǎn)觸發(fā)

(2)interval: 固定時(shí)間間隔觸發(fā)

(3)cron: 在特定時(shí)間周期性地觸發(fā)

2、任務(wù)儲(chǔ)存器 job stores：用于存放任務(wù)，把任務(wù)存放在內(nèi)存(為默認(rèn)MemoryJobStore)或數(shù)據(jù)庫(kù)中。

3、執(zhí)行器 executors： 執(zhí)行器是將任務(wù)提交到線程池或進(jìn)程池中運(yùn)行，當(dāng)任務(wù)完成時(shí)，執(zhí)行器通知調(diào)度器觸發(fā)相應(yīng)的事件。

4、調(diào)度器 schedulers： 把上方三個(gè)組件作為參數(shù)，通過(guò)創(chuàng)建調(diào)度器實(shí)例來(lái)運(yùn)行

根據(jù)開(kāi)發(fā)需求選擇相應(yīng)的組件，下面是不同的調(diào)度器組件：

BlockingScheduler 阻塞式調(diào)度器：適用于只跑調(diào)度器的程序。

BackgroundScheduler 后臺(tái)調(diào)度器：適用于非阻塞的情況，調(diào)度器會(huì)在后臺(tái)獨(dú)立運(yùn)行。

AsyncIOScheduler AsyncIO調(diào)度器，適用于應(yīng)用使用AsnycIO的情況。

GeventScheduler Gevent調(diào)度器，適用于應(yīng)用通過(guò)Gevent的情況。

TornadoScheduler Tornado調(diào)度器，適用于構(gòu)建Tornado應(yīng)用。

TwistedScheduler Twisted調(diào)度器，適用于構(gòu)建Twisted應(yīng)用。

QtScheduler Qt調(diào)度器，適用于構(gòu)建Qt應(yīng)用。

三、使用步驟

1、新建一個(gè)調(diào)度器schedulers

2、添加調(diào)度任務(wù)

3、運(yùn)行調(diào)度任務(wù)

四、使用實(shí)例

1、觸發(fā)器date

特定的時(shí)間點(diǎn)觸發(fā)，只執(zhí)行一次。參數(shù)如下：

參數(shù)

說(shuō)明

run_date (datetime 或 str)

作業(yè)的運(yùn)行日期或時(shí)間

timezone (datetime.tzinfo 或 str)

指定時(shí)區(qū)

使用例子：

from datetime import datetime

from datetime import date

from apscheduler.schedulers.blocking import BlockingScheduler

def job(text):

print(text)

scheduler = BlockingScheduler()

# 在 2019-8-30 運(yùn)行一次 job 方法

scheduler.add_job(job, 'date', run_date=date(2019, 8, 30), args=['text1'])

# 在 2019-8-30 01:00:00 運(yùn)行一次 job 方法

scheduler.add_job(job, 'date', run_date=datetime(2019, 8, 30, 1, 0, 0), args=['text2'])

# 在 2019-8-30 01:00:01 運(yùn)行一次 job 方法

scheduler.add_job(job, 'date', run_date='2019-8-30 01:00:00', args=['text3'])

scheduler.start()

2、觸發(fā)器interval

固定時(shí)間間隔觸發(fā)。參數(shù)如下：

參數(shù)

說(shuō)明

weeks (int)

間隔幾周

days (int)

間隔幾天

hours (int)

間隔幾小時(shí)

minutes (int)

間隔幾分鐘

seconds (int)

間隔多少秒

start_date (datetime 或 str)

開(kāi)始日期

end_date (datetime 或 str)

結(jié)束日期

timezone (datetime.tzinfo 或str)

使用例子：

import time

from apscheduler.schedulers.blocking import BlockingScheduler

def job(text):

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

print('{} --- {}'.format(text, t))

scheduler = BlockingScheduler()

# 每隔 1分鐘 運(yùn)行一次 job 方法

scheduler.add_job(job, 'interval', minutes=1, args=['job1'])

# 在 2019-08-29 22:15:00至2019-08-29 22:17:00期間，每隔1分30秒 運(yùn)行一次 job 方法

scheduler.add_job(job, 'interval', minutes=1, seconds = 30, start_date='2019-08-29 22:15:00', end_date='2019-08-29 22:17:00', args=['job2'])

scheduler.start()

'''

運(yùn)行結(jié)果：

job2 --- 2019-08-29 22:15:00

job1 --- 2019-08-29 22:15:46

job2 --- 2019-08-29 22:16:30

job1 --- 2019-08-29 22:16:46

job1 --- 2019-08-29 22:17:46

...余下省略...

'''

3、觸發(fā)器cron

在特定時(shí)間周期性地觸發(fā)。參數(shù)如下：

參數(shù)

說(shuō)明

year (int 或 str)

年，4位數(shù)字

month (int 或 str)

月 (范圍1-12)

day (int 或 str)

日 (范圍1-31)

week (int 或 str)

周 (范圍1-53)

day_of_week (int 或 str)

周內(nèi)第幾天或者星期幾 (范圍0-6 或者 mon,tue,wed,thu,fri,sat,sun)

hour (int 或 str)

時(shí) (范圍0-23)

minute (int 或 str)

分 (范圍0-59)

second (int 或 str)

秒 (范圍0-59)

start_date (datetime 或 str)

最早開(kāi)始日期(包含)

end_date (datetime 或 str)

最晚結(jié)束時(shí)間(包含)

timezone (datetime.tzinfo 或str)

指定時(shí)區(qū)

這些參數(shù)支持算數(shù)表達(dá)式，取值格式有如下：

使用例子：

import time

from apscheduler.schedulers.blocking import BlockingScheduler

def job(text):

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

print('{} --- {}'.format(text, t))

scheduler = BlockingScheduler()

# 在每天22點(diǎn)，每隔 1分鐘 運(yùn)行一次 job 方法

scheduler.add_job(job, 'cron', hour=22, minute='*/1', args=['job1'])

# 在每天22和23點(diǎn)的25分，運(yùn)行一次 job 方法

scheduler.add_job(job, 'cron', hour='22-23', minute='25', args=['job2'])

scheduler.start()

'''

運(yùn)行結(jié)果：

job1 --- 2019-08-29 22:25:00

job2 --- 2019-08-29 22:25:00

job1 --- 2019-08-29 22:26:00

job1 --- 2019-08-29 22:27:00

...余下省略...

'''

4、通過(guò)裝飾器scheduled_job()添加方法

添加任務(wù)的方法有兩種：

(1)通過(guò)調(diào)用add_job()---見(jiàn)上面1至3代碼

(2)通過(guò)裝飾器scheduled_job()：

第一種方法是最常用的方法。第二種方法主要是方便地聲明在應(yīng)用程序運(yùn)行時(shí)不會(huì)更改的任務(wù)。該 add_job()方法返回一個(gè)apscheduler.job.Job實(shí)例，可以使用該實(shí)例稍后修改或刪除該任務(wù)。

import time

from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

@scheduler.scheduled_job('interval', seconds=5)

def job1():

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

print('job1 --- {}'.format(t))

@scheduler.scheduled_job('cron', second='*/7')

def job2():

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

print('job2 --- {}'.format(t))

scheduler.start()

'''

運(yùn)行結(jié)果：

job2 --- 2019-08-29 22:36:35

job1 --- 2019-08-29 22:36:37

job2 --- 2019-08-29 22:36:42

job1 --- 2019-08-29 22:36:42

job1 --- 2019-08-29 22:36:47

job2 --- 2019-08-29 22:36:49

...余下省略...

'''

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

總結(jié)

以上是生活随笔為你收集整理的python apscheduler一次只有一个job_Python使用APScheduler实现定时任务过程解析的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

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