Python 调度相关包的使用

schedule

使用起来比较简单的一个包

安装：

代码语言：bash复制

pip install schedule

具体用法：

代码语言：python代码运行次数：0复制

import schedule

# add schedule job
schedule.every(10).seconds.do(lambda: print("running"))

# run scheduler
while True:
    schedule.run_pending()
    time.sleep(1)

运行带有参数的 job

代码语言：python代码运行次数：0复制

def func(name: str):
    print(f"My name is {name}")
    
schedule.every(5).seconds.do(func, name="Tom")
while True:
    schedule.run_pending()
    time.sleep(1)

Apscheduler

一个功能更为完整的包

安装：

代码语言：bash复制

pip install apscheduler

一些基本概念：

Triggers：任务触发逻辑

cron：cron 格式触发
interval：固定时间间隔触发
date：在某固定日期触发一次
combine：组合条件触发

Scheduler

BlockingScheduler：阻塞式，当程序只运行这个 scheduler 时使用
BackgroundScheduler：调度器在后台运行

Executor

ThreadPoolExecutor：默认使用多线程执行器
ProcessPoolExecutor：如果是 CPU 密集型任务可以使用多进程执行器

Job store：如果任务调度信息存在内存中，当程序退出后会丢失，可以其他存储器进行持久化存储

MemoryJobStore：默认使用内存存储
SQLAlchemyJobStore
MongoDBJobStore
etc.

创建 scheduler

代码语言：python代码运行次数：0复制

# 创建 BlockingScheduler
from apscheduler.schedulers.blocking import BlockingScheduler
scheduler = BlockingScheduler()

# 创建 BackgroundScheduler
from apscheduler.schedulers.background import BackgroundScheduler
scheduler = BackgroundScheduler()

# 自定义 job store、executor、job defaults、time zone
import pytz
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor

scheduler = BackgroundScheduler(
    jobstores={'mongo': MongoDBJobStore()},
    executors={'processpool': ProcessPoolExecutor(5)},
    job_defaults = {
        'coalesce': False,
        'max_instances': 3
    },
    timezone=pytz.utc
)

新增任务

代码语言：python代码运行次数：0复制

from datetime import datetime
def tick():
    print(datetime.now().strftime('%Y-%m-%d %H:%M:%S'))

# add job with cron trigger
scheduler.add_job(tick, "cron", second=0) # every minute at second 0

先创建 triggers 再创建任务

代码语言：python代码运行次数：0复制

from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.cron import CronTrigger

c_trigger = CronTrigger(second=0) # every minute at second 0
i_trigger = IntervalTrigger(seconds=10) # every 10 seconds

scheduler.add_job(tick, c_trigger)
scheduler.add_job(tick, i_trigger)

带参数任务

代码语言：python代码运行次数：0复制

def whoami(name: str, age: int):
    print(f"My name is {name}, age is {age}")

scheduler.add_job(whoami, i_trigger, args=("Tom",))
# or
scheduler.add_job(whoami, i_trigger, kwargs={"name":"Tom", "age":13})

启动调度器

代码语言：python代码运行次数：0复制

# start
scheduler.start()

# shutdown
scheduler.shutdown()

scheduler python

0 人点赞