一、APScheduler 簡(jiǎn)介

1. schedule：第三方模塊，該模塊適合比較輕量級(jí)的一些調(diào)度任務(wù)，但卻不適用于復(fù)雜時(shí)間的調(diào)度
2. APScheduler：第三方定時(shí)任務(wù)框架，是對(duì) Java 第三方定時(shí)任務(wù)框架 Quartz 的模仿與移植，能提供比 schedule 更復(fù)雜的應(yīng)用場(chǎng)景，并且各種組件都是模塊化，易于使用與二次開發(fā)。
3. Celery Beat：屬于 celery 這分布式任務(wù)隊(duì)列第三方庫(kù)下的一個(gè)定時(shí)任務(wù)組件，如果使用需要配合 RabbitMQ 或 Redis 這類的消息隊(duì)列套件，需要花費(fèi)一定的時(shí)間在環(huán)境搭建上，但在高版本中已經(jīng)不支持 Windows。

二、APScheduler 概念與組件

• 觸發(fā)器（trigger）
• 任務(wù)持久化（job stores）
• 執(zhí)行器（executor）
• 調(diào)度器（scheduler）

1.觸發(fā)器（trigger）

• date：日期觸發(fā)器。日期觸發(fā)器主要是在某一日期時(shí)間點(diǎn)上運(yùn)行任務(wù)時(shí)調(diào)用，是 APScheduler 里面最簡(jiǎn)單的一種觸發(fā)器。所以通常也適用于一次性的任務(wù)或作業(yè)調(diào)度。
• interval：間隔觸發(fā)器。間隔觸發(fā)器是在日期觸發(fā)器基礎(chǔ)上擴(kuò)展了對(duì)時(shí)間部分，比如時(shí)、分、秒、天、周這幾個(gè)部分的設(shè)定。是我們用以對(duì)重復(fù)性任務(wù)進(jìn)行設(shè)定或調(diào)度的一個(gè)常用調(diào)度器。設(shè)定了時(shí)間部分之后，從起始日期開始（默認(rèn)是當(dāng)前）會(huì)按照設(shè)定的時(shí)間去執(zhí)行任務(wù)。
• cron：cron 表達(dá)式觸發(fā)器。cron 表達(dá)式觸發(fā)器就等價(jià)于我們 Linux 上的 crontab，它主要用于更復(fù)雜的日期時(shí)間進(jìn)行設(shè)定。但需要注意的是，APScheduler 不支持 6 位及以上的 cron 表達(dá)式，最多只支持到 5 位。

2.任務(wù)持久化（job stores）

• sqlalchemy 形式的數(shù)據(jù)庫(kù)，這里就主要是指各種傳統(tǒng)的關(guān)系型數(shù)據(jù)庫(kù)，如 MySQL、PostgreSQL、SQLite 等。
• mongodb 非結(jié)構(gòu)化的 Mongodb 數(shù)據(jù)庫(kù)，該類型數(shù)據(jù)庫(kù)經(jīng)常用于對(duì)非結(jié)構(gòu)化或版結(jié)構(gòu)化數(shù)據(jù)的存儲(chǔ)或操作，如 JSON。
• redis 內(nèi)存數(shù)據(jù)庫(kù)，通常用作數(shù)據(jù)緩存來使用，當(dāng)然通過一些主從復(fù)制等方式也能實(shí)現(xiàn)當(dāng)中數(shù)據(jù)的持久化或保存。

3.執(zhí)行器（executor）

4.調(diào)度器（scheduler）

• BlockingScheduler：阻塞調(diào)度器，當(dāng)程序中沒有任何存在主進(jìn)程之中運(yùn)行東西時(shí)，就則使用該調(diào)度器。
• BackgroundScheduler：后臺(tái)調(diào)度器，在不使用后面任何的調(diào)度器且希望在應(yīng)用程序內(nèi)部運(yùn)行時(shí)的后臺(tái)啟動(dòng)時(shí)才進(jìn)行使用，如當(dāng)前你已經(jīng)開啟了一個(gè) Django 或 Flask 服務(wù)。
• AsyncIOScheduler：AsyncIO 調(diào)度器，如果代碼是通過 asyncio 模塊進(jìn)行異步操作，使用該調(diào)度器。
• GeventScheduler：Gevent 調(diào)度器，如果代碼是通過 gevent 模塊進(jìn)行協(xié)程操作，使用該調(diào)度器
• TornadoScheduler：Tornado 調(diào)度器，在 Tornado 框架中使用
• TwistedScheduler：Twisted 調(diào)度器，在基于 Twisted 的框架或應(yīng)用程序中使用
• QtScheduler：Qt 調(diào)度器，在構(gòu)建 Qt 應(yīng)用中進(jìn)行使用。

1. 設(shè)定調(diào)度器（scheduler）用以對(duì)任務(wù)的調(diào)度與安排進(jìn)行全局統(tǒng)籌
2. 對(duì)相應(yīng)的函數(shù)或方法上設(shè)定相應(yīng)的觸發(fā)器（trigger），并添加到調(diào)度器中
3. 如有任務(wù)持久化（job stores）需要?jiǎng)t需要設(shè)定對(duì)應(yīng)的持久化層，否則默認(rèn)使用內(nèi)存存儲(chǔ)任務(wù)
4. 當(dāng)觸發(fā)器被觸發(fā)時(shí)，就將任務(wù)交由執(zhí)行器（executor）進(jìn)行執(zhí)行

三、APScheduler 快速上手

1.選擇對(duì)應(yīng)的 scheduler

#?main.py

from?apscheduler.schedulers.blocking?import?BlockingScheduler

scheduler?=?BlockingScheduler()

2.配置 scheduler

#?main.py
from?datetime?import?datetime

from?apscheduler.executors.pool?import?ThreadPoolExecutor
from?apscheduler.jobstores.sqlalchemy?import?SQLAlchemyJobStore
from?apscheduler.schedulers.blocking?import?BlockingScheduler

#?任務(wù)持久化?使用?SQLite
jobstores?=?{
????'default':?SQLAlchemyJobStore(url?=?'sqlite:///jobs.db')
}
#?執(zhí)行器配置
executors?=?{
????'default':?ThreadPoolExecutor(20),
}
#?關(guān)于?Job?的相關(guān)配置，見官方文檔?API
job_defaults?=?{
????'coalesce':?False,
????'next_run_time':?datetime.now()
}
scheduler?=?BlockingScheduler(
??jobstores?=?jobstores,
??executors?=?executors,
??job_defaults?=?job_defaults,
??timezone?=?'Asia/Shanghai'
)

scheduler?=?BlockingScheduler()
scheduler.configure(jobstores=jobstores,?executors=executors,?job_defaults=job_defaults,?timezone='Asia/Shanghai')

3.添加并執(zhí)行你的任務(wù)

from?datetime?import?datetime

def?now(trigger):
????print(f"trigger:{trigger}?->?{datetime.now()}")

if?__name__?==?'__main__':
????scheduler.add_job(now,?trigger?=?"interval",?args?=?("interval",),?seconds?=?5)
????scheduler.start()

trigger:interval?->?2021-01-16?21:19:43.356674
trigger:interval?->?2021-01-16?21:19:46.679849
trigger:interval?->?2021-01-16?21:19:48.356595

@scheduler.scheduled_job(trigger?=?"interval",?args?=?("interval",),?seconds?=?5)
def?now(trigger):
????print(f"trigger:{trigger}?->?{datetime.now()}")

if?__name__?==?'__main__':
????scheduler.start()

四、將 APScheduler 集成到 Web 項(xiàng)目中

temp-scheduler
├──?config.py???????#?配置項(xiàng)
├──?main.py?????????#?API?文件
└──?scheduler.py????#?APScheduler?相關(guān)設(shè)置

1.安裝依賴

這里我們需要的依賴不多，只需要簡(jiǎn)單幾個(gè)即可：

pip?install?fastapi?apscheduler?sqlalchemy?uvicorn

2.配置項(xiàng)

from?apscheduler.executors.pool?import?ThreadPoolExecutor
from?apscheduler.jobstores.sqlalchemy?import?SQLAlchemyJobStore
from?apscheduler.schedulers.blocking?import?BlockingScheduler

class?SchedulerConfig:

????JOBSTORES?=?{"default":?SQLAlchemyJobStore(url="sqlite:///job.db")}
????EXECUTORS?=?{"default":?ThreadPoolExecutor(20)}
????JOB_DEFAULTS?=?{"coalesce":?False}

????@classmethod
????def?to_dict(cls):
????????return?{
????????????"jobstores":?cls.JOBSTORES,
????????????"executors":?cls.EXECUTORS,
????????????"job_defaults":?cls.JOB_DEFAULTS,
????????}

3.Scheduler 相關(guān)設(shè)置

import?logging
from?datetime?import?datetime

from?apscheduler.schedulers.background?import?BackgroundScheduler

from?config?import?SchedulerConfig

scheduler?=?BackgroundScheduler()
logger?=?logging.getLogger(__name__)

def?init_scheduler()?->?None:
????#?config?scheduler
????scheduler.configure(**SchedulerConfig.to_dict())

????logger.info("scheduler?is?running...")

????#?schedule?test
????scheduler.add_job(
????????func=mytask,
????????trigger="date",
????????args=("APScheduler?Initialize.",),
????????next_run_time=datetime.now(),
????)
????scheduler.start()

def?mytask(message:?str)?->?None:
????print(f"[{datetime.now()}]?message:?{message}")

• init_scheduler() 方法主要用于在 API 服務(wù)啟動(dòng)時(shí)被調(diào)用，然后對(duì) scheduler 對(duì)象的配置以及測(cè)試
• mytask() 則是我們要定期執(zhí)行的任務(wù)，后續(xù)我們可以通過 APScheduler 提供的方法來自行添加任務(wù)

4.API 設(shè)置

import?logging
import?uuid
from?datetime?import?datetime
from?typing?import?Any,?Dict,?Optional,?Sequence,?Union

from?fastapi?import?FastAPI
from?pydantic?import?BaseModel

from?scheduler?import?init_scheduler,?mytask,?scheduler

logger?=?logging.getLogger(__name__)

app?=?FastAPI(title="APScheduler?API")
app.add_event_handler("startup",?init_scheduler)

class?Job(BaseModel):
????id:?Union[int,?str,?uuid.UUID]
????name:?Optional[str]?=?None
????func:?Optional[str]?=?None
????args:?Optional[Sequence[Optional[str]]]?=?None
????kwargs:?Optional[Dict[str,?Any]]?=?None
????executor:?Optional[str]?=?None
????misfire_grace_time:?Optional[str]?=?None
????coalesce:?Optional[bool]?=?None
????max_instances:?Optional[int]?=?None
????next_run_time:?Optional[Union[str,?datetime]]?=?None

@app.post("/add")
def?add_job(
????message:?str,
????trigger:?str,
????trigger_args:?Optional[dict],
????id:?Union[str,?int,?uuid.UUID],
):
????try:
????????scheduler.add_job(
????????????func=mytask,
????????????trigger=trigger,
????????????kwargs={"message":?message},
????????????id=id,
????????????**trigger_args,
????????)
????except?Exception?as?e:
????????logger.exception(e.args)
????????return?{"status_code":?0,?"message":?"添加失敗"}
????return?{"status_code":?1,?"message":?"添加成功"}

@app.delete("/delete/{id}")
def?delete_job(id:?Union[str,?int,?uuid.UUID]):
????"""delete?exist?job?by?id"""
????try:
????????scheduler.remove_job(job_id=id)
????except?Exception:
????????return?dict(
????????????message="刪除失敗",
????????????status_code=0,
????????)
????return?dict(
????????message="刪除成功",
????????status_code=1,
????)

@app.put("/reschedule/{id}")
def?reschedule_job(
????id:?Union[str,?int,?uuid.UUID],?trigger:?str,?trigger_args:?Optional[dict]
):
????try:
????????scheduler.reschedule_job(job_id=id,?trigger=trigger,?**trigger_args)
????except?Exception?as?e:
????????logger.exception(e.args)
????????return?dict(
????????????message="修改失敗",
????????????status_code=0,
????????)
????return?dict(
????????message="修改成功",
????????status_code=1,
????)

@app.get("/job")
def?get_all_jobs():
????jobs?=?None
????try:
????????job_list?=?scheduler.get_jobs()
????????if?job_list:
????????????jobs?=?[Job(**task.__getstate__())?for?task?in?job_list]
????except?Exception?as?e:
????????logger.exception(e.args)
????????return?dict(
????????????message="查詢失敗",
????????????status_code=0,
????????????jobs=jobs,
????????)
????return?dict(
????????message="查詢成功",
????????status_code=1,
????????jobs=jobs,
????)

@app.get("/job/{id}")
def?get_job_by_id(id:?Union[int,?str,?uuid.UUID]):
????jobs?=?[]
????try:
????????job?=?scheduler.get_job(job_id=id)
????????if?job:
????????????jobs?=?[Job(**job.__getstate__())]
????except?Exception?as?e:
????????logger.exception(e.args)
????????return?dict(
????????????message="查詢失敗",
????????????status_code=0,
????????????jobs=jobs,
????????)
????return?dict(
????????message="查詢成功",
????????status_code=1,
????????jobs=jobs,
????)

1. FastAPI 對(duì)象 app 的初始化。這里用到的 add_event_handler() 方法就有點(diǎn)像 Flask 中的 before_first_request，會(huì)在 Web 服務(wù)請(qǐng)求伊始進(jìn)行操作，理解為初始化相關(guān)的操作即可。
2. API 接口路由。路由通過 app 對(duì)象下的對(duì)應(yīng) HTTP 方法來實(shí)現(xiàn)，如 GET、POST、PUT 等。這里的裝飾器用法其實(shí)也和 Flask 很類似，就不多贅述。
3. scheduler 對(duì)象的增刪改查。從 scheduler.py 模塊中引入我們創(chuàng)建好的 scheduler 對(duì)象之后就可以直接用來做增刪改查的操作：
1. 增：使用 add_job() 方法，其主要的參數(shù)是要運(yùn)行的函數(shù)（或方法）、觸發(fā)器以及觸發(fā)器參數(shù)等
2. 刪：使用 delete_job() 方法，我們需要傳入一個(gè)對(duì)應(yīng)任務(wù)的 id 參數(shù)，用以能夠查找到對(duì)應(yīng)的任務(wù)
3. 改：使用 reschedule_job() 方法，這里也需要一個(gè)對(duì)應(yīng)任務(wù)的 id 參數(shù)，以及需要重新修改的觸發(fā)器及其參數(shù)
4. 查：使用 get_jobs() 和 get_job() 兩個(gè)方法，前者是直接獲取到當(dāng)前調(diào)度的所有任務(wù)，返回的是一個(gè)包含了 APScheduler.job.Job 對(duì)象的列表，而后者是通過 id 參數(shù)來查找對(duì)應(yīng)的任務(wù)對(duì)象；這里我通過底層源碼使用 __getstate__() 來獲取到任務(wù)的相關(guān)信息，這些信息我們通過事先設(shè)定好的 Job 對(duì)象來對(duì)其進(jìn)行序列化，最后將信息從接口中返回。

5.運(yùn)行

uvicorn?main:app?

五、結(jié)尾