Python APSchedule实现定时任务的速查手册

pip install apscheduler

from apscheduler.schedulers.blocking import BlockingScheduler
import datetime

# 定义一个简单的任务函数
def my_job():
    print(f"Job executed at {datetime.datetime.now()}")

# 创建调度器
scheduler = BlockingScheduler()

# 添加任务，每隔 5 秒执行一次
scheduler.add_job(my_job, 'interval', seconds=5)

# 启动调度器
scheduler.start()

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
import datetime

# 定义一个任务函数
def my_cron_job():
    print(f"Cron job executed at {datetime.datetime.now()}")

# 创建调度器
scheduler = BlockingScheduler()

# 添加任务，每天早上 8 点执行
cron_trigger = CronTrigger(hour=8, minute=0, second=0)
scheduler.add_job(my_cron_job, cron_trigger)

# 启动调度器
scheduler.start()

from apscheduler.schedulers.blocking import BlockingScheduler
import datetime

# 定义一个任务函数
def my_job():
    print(f"Job executed at {datetime.datetime.now()}")

# 创建调度器
scheduler = BlockingScheduler()

# 添加任务，任务 ID 为 'job1'
job = scheduler.add_job(my_job, 'interval', seconds=5, id='job1')

# 启动调度器
try:
    scheduler.start()
except (KeyboardInterrupt, SystemExit):
    # 暂停任务
    job.pause()
    print("Task paused")
    # 重新启动任务
    job.resume()
    print("Task resumed")
    # 最终取消任务
    scheduler.remove_job('job1')
    print("Task removed")

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
import datetime

# 创建调度器，使用 SQLAlchemyJobStore 进行任务持久化
jobstores = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.db')
}
scheduler = BlockingScheduler(jobstores=jobstores)

# 定义一个任务函数
def my_persistent_job():
    print(f"Persistent job executed at {datetime.datetime.now()}")

# 添加任务，每隔 5 秒执行一次
job = scheduler.add_job(my_persistent_job, 'interval', seconds=5, id='job1')

# 启动调度器
scheduler.start()

from apscheduler.schedulers.blocking import BlockingScheduler
import datetime

# 定义一个任务函数
def my_job():
    print(f"Job executed at {datetime.datetime.now()}")

# 创建调度器
scheduler = BlockingScheduler()

# 添加任务，每隔 5 秒执行一次，允许最多 2 个实例同时运行，误点 10 秒内执行
job = scheduler.add_job(my_job, 'interval', seconds=5, id='job1', max_instances=2, misfire_grace_time=10)

# 启动调度器
scheduler.start()

from apscheduler.schedulers.background import BackgroundScheduler
import datetime
import time

# 定义一个任务函数
def my_job():
    print(f"Job executed at {datetime.datetime.now()}")

# 创建调度器
scheduler = BackgroundScheduler()

# 添加任务，每隔 5 秒执行一次
scheduler.add_job(my_job, 'interval', seconds=5)

# 启动调度器
scheduler.start()

# 主线程继续执行其他任务
try:
    while True:
        time.sleep(2)
        print("Main thread is running...")
except (KeyboardInterrupt, SystemExit):
    # 清理资源
    scheduler.shutdown()