Python中定时任务APScheduler框架使用教程

pip install apscheduler

from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler

def once_task():
    print(f"一次性任务执行：{datetime.now()}")

scheduler = BlockingScheduler()
# 指定时间执行一次
scheduler.add_job(once_task, "date", run_date="2026-12-31 18:00:00")
scheduler.start()

from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler

def loop_task():
    print(f"间隔轮询任务：{datetime.now()}")

scheduler = BlockingScheduler()
# 每 3 秒执行一次
scheduler.add_job(loop_task, "interval", seconds=3)
# 常用：minutes=5 / hours=1 / days=1
scheduler.start()

from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler

def cron_spider_task():
    print(f"定时爬虫执行成功：{datetime.now()}")

scheduler = BlockingScheduler()

# 每天 8/10/12/15/17/22 整点执行
scheduler.add_job(
    func=cron_spider_task,
    trigger="cron",
    hour="8,10,12,15,17,22",
    minute="0",
    second="0",
    id="spider_cron_task"
)

scheduler.start()

from apscheduler.schedulers.blocking import BlockingScheduler

def task_demo(name, mode="定时执行"):
    print(f"任务：{name}，执行模式：{mode}")

scheduler = BlockingScheduler()
scheduler.add_job(
    func=task_demo,
    trigger="cron",
    hour=8,
    minute=0,
    args=("SMS新闻爬虫",),
    kwargs={"mode": "每日定点轮询"},
    id="task_param_demo"
)
scheduler.start()

from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler

def spider_task():
    print("爬虫任务执行")

scheduler = BlockingScheduler()
scheduler.add_job(
    func=spider_task,
    trigger="cron",
    hour="8,22",
    minute="0",
    next_run_time=datetime.now()
)
scheduler.start()

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor

# 自定义并发池
executors = {
    "default": ThreadPoolExecutor(100),   # IO爬虫用线程池
    "processpool": ProcessPoolExecutor(1) # 计算任务用进程池
}

scheduler = BlockingScheduler(executors=executors)

# 根据id删除任务
scheduler.remove_job("task_id")

# 暂停任务
scheduler.pause_job("task_id")

# 恢复任务
scheduler.resume_job("task_id")

# 查看所有任务
print(scheduler.get_jobs())

import traceback
from functools import wraps
from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor

# 任务日志装饰器
def task_logger(task_name):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            print(f"[{datetime.now()}]【{task_name}】任务开始")
            try:
                res = func(*args, **kwargs)
                print(f"[{datetime.now()}]【{task_name}】任务执行成功")
                return res
            except Exception as e:
                err = traceback.format_exc()
                print(f"[{datetime.now()}]【{task_name}】任务异常：{repr(e)}\n{err}")
                raise
        return wrapper
    return decorator

# 业务任务
@task_logger("SMS新闻爬虫")
def spider_task():
    # 此处替换为你的爬虫逻辑
    pass

if __name__ == "__main__":
    # 并发配置
    executors = {
        "default": ThreadPoolExecutor(100),
        "processpool": ProcessPoolExecutor(1)
    }
    scheduler = BlockingScheduler(executors=executors)

    # 添加定时任务
    scheduler.add_job(
        func=spider_task,
        trigger="cron",
        hour="8,10,12,15,17,22",
        minute="0",
        second="0",
        id="sms_spider_task",
        next_run_time=datetime.now()
    )

    print("定时调度器启动成功")
    scheduler.start()