快捷導(dǎo)航

apschedule的定時(shí)任務(wù)實(shí)例(python)

更新時(shí)間：2026年01月07日 16:03:00 作者：小徐敲java

文章介紹了四種配置定時(shí)任務(wù)的方法,包括直接代碼配置、YAML配置文件管理、裝飾器方式自動(dòng)注冊(cè)任務(wù)以及關(guān)閉定時(shí)任務(wù)的方法

1：方法一：直接代碼配置定時(shí)任務(wù)

使用APScheduler的BackgroundScheduler直接配置多個(gè)定時(shí)任務(wù)，通過環(huán)境變量控制執(zhí)行間隔

import os
from apscheduler.schedulers.background import BackgroundScheduler

def init_scheduler():
    """初始化多個(gè)定時(shí)任務(wù)"""
    scheduler = BackgroundScheduler(timezone="Asia/Shanghai")
    
    # 定義任務(wù)配置
    tasks_config = [
        {
            'func': sync_analysis_s7_data,
            'name': '同步AnalysisS7數(shù)據(jù)',
            'interval': int(os.getenv("TASK_INTERVAL_S7", 10)),
            'id': 'sync_analysis_s7'
        },
        {
            'func': sync_user_data,
            'name': '同步用戶數(shù)據(jù)',
            'interval': int(os.getenv("TASK_INTERVAL_USER", 30)),
            'id': 'sync_user_data'
        },
        {
            'func': generate_report,
            'name': '生成報(bào)表',
            'interval': int(os.getenv("TASK_INTERVAL_REPORT", 3600)),
            'id': 'generate_report'
        }
    ]
    
    # 動(dòng)態(tài)添加任務(wù)
    for task in tasks_config:
        scheduler.add_job(
            func=task['func'],
            trigger="interval",
            seconds=task['interval'],
            id=task['id'],
            name=task['name'],
            replace_existing=True
        )
        print(f"任務(wù)已添加：{task['name']}，間隔：{task['interval']}秒")
        logger.info(f"任務(wù)已添加：{task['name']}，間隔：{task['interval']}秒")
    
    scheduler.start()
    print("所有定時(shí)任務(wù)已啟動(dòng)")
    logger.info("所有定時(shí)任務(wù)已啟動(dòng)")
    return scheduler

2：方法二：YAML配置文件管理

建一個(gè)配置文件 tasks.yaml

通過YAML文件定義任務(wù)配置，支持interval和cron兩種觸發(fā)器類型

tasks:
  - name: "同步AnalysisS7數(shù)據(jù)"
    function: "sync_analysis_s7_data"
    interval_seconds: 10
    id: "sync_analysis_s7"
  
  - name: "同步用戶數(shù)據(jù)"
    function: "sync_user_data"
    interval_seconds: 30
    id: "sync_user_data"
    
  - name: "生成報(bào)表"
    function: "generate_report"
    interval_seconds: 3600
    id: "generate_report"
    
  - name: "清理日志"
    function: "clean_logs"
    cron: "0 2 * * *"  # 每天凌晨2點(diǎn)執(zhí)行
    id: "clean_logs"

對(duì)應(yīng)的實(shí)現(xiàn)代碼

import yaml
from apscheduler.schedulers.background import BackgroundScheduler

def load_tasks_from_yaml(config_path):
    """從YAML配置文件加載任務(wù)"""
    with open(config_path, 'r', encoding='utf-8') as f:
        config = yaml.safe_load(f)
    return config['tasks']

def get_function_by_name(name):
    """根據(jù)名稱獲取對(duì)應(yīng)的函數(shù)對(duì)象"""
    functions = {
        'sync_analysis_s7_data': sync_analysis_s7_data,
        'sync_user_data': sync_user_data,
        'generate_report': generate_report,
        'clean_logs': clean_logs
    }
    return functions.get(name)

def init_scheduler():
    """基于配置文件初始化定時(shí)任務(wù)"""
    scheduler = BackgroundScheduler(timezone="Asia/Shanghai")
    
    # 加載任務(wù)配置
    tasks = load_tasks_from_yaml('tasks.yaml')
    
    # 添加任務(wù)
    for task in tasks:
        func = get_function_by_name(task['function'])
        if not func:
            logger.warning(f"未找到函數(shù): {task['function']}")
            continue
            
        # 支持interval和cron兩種觸發(fā)器
        if 'interval_seconds' in task:
            scheduler.add_job(
                func=func,
                trigger="interval",
                seconds=task['interval_seconds'],
                id=task['id'],
                name=task['name'],
                replace_existing=True
            )
            print(f"間隔任務(wù)已添加：{task['name']}，間隔：{task['interval_seconds']}秒")
            
        elif 'cron' in task:
            cron_parts = task['cron'].split()
            scheduler.add_job(
                func=func,
                trigger="cron",
                minute=cron_parts[0],
                hour=cron_parts[1],
                day=cron_parts[2],
                month=cron_parts[3],
                day_of_week=cron_parts[4],
                id=task['id'],
                name=task['name'],
                replace_existing=True
            )
            print(f"Cron任務(wù)已添加：{task['name']}，調(diào)度：{task['cron']}")
    
    scheduler.start()
    print("所有定時(shí)任務(wù)已啟動(dòng)")
    return scheduler

3：裝飾器方式自動(dòng)注冊(cè)任務(wù)

使用裝飾器自動(dòng)收集任務(wù)配置，簡(jiǎn)化任務(wù)注冊(cè)過程

from functools import wraps
from apscheduler.schedulers.background import BackgroundScheduler

# 全局任務(wù)注冊(cè)表
_task_registry = []

def scheduled_task(name, interval_seconds=None, cron=None, task_id=None):
    """裝飾器：注冊(cè)定時(shí)任務(wù)"""
    def decorator(func):
        _task_registry.append({
            'func': func,
            'name': name,
            'interval_seconds': interval_seconds,
            'cron': cron,
            'id': task_id or func.__name__
        })
        return func
    return decorator

# 使用裝飾器定義任務(wù)
@scheduled_task(name="同步AnalysisS7數(shù)據(jù)", interval_seconds=10)
def sync_analysis_s7_data():
    pass

@scheduled_task(name="同步用戶數(shù)據(jù)", interval_seconds=30)
def sync_user_data():
    pass

@scheduled_task(name="每日?qǐng)?bào)告", cron="0 9 * * *")  # 每天9點(diǎn)執(zhí)行
def daily_report():
    pass

def init_scheduler():
    """初始化所有裝飾器注冊(cè)的任務(wù)"""
    scheduler = BackgroundScheduler(timezone="Asia/Shanghai")
    
    for task_info in _task_registry:
        kwargs = {
            'func': task_info['func'],
            'id': task_info['id'],
            'name': task_info['name'],
            'replace_existing': True
        }
        
        # 根據(jù)配置選擇觸發(fā)器類型
        if task_info['interval_seconds']:
            kwargs.update({
                'trigger': "interval",
                'seconds': task_info['interval_seconds']
            })
        elif task_info['cron']:
            cron_parts = task_info['cron'].split()
            kwargs.update({
                'trigger': "cron",
                'minute': cron_parts[0],
                'hour': cron_parts[1],
                'day': cron_parts[2],
                'month': cron_parts[3],
                'day_of_week': cron_parts[4]
            })
        
        scheduler.add_job(**kwargs)
        print(f"任務(wù)已添加：{task_info['name']}")
    
    scheduler.start()
    print("所有定時(shí)任務(wù)已啟動(dòng)")
    return scheduler

4：注意最終記得要關(guān)閉定時(shí)任務(wù)

4-1：方法一

def signal_handler(signum, frame):
    """信號(hào)處理函數(shù)"""
    print("接收到關(guān)閉信號(hào)，正在停止定時(shí)任務(wù)...")
    scheduler.shutdown()
    print("定時(shí)任務(wù)已停止")
    sys.exit(0)

# 注冊(cè)信號(hào)處理器
signal.signal(signal.SIGINT, signal_handler)   # Ctrl+C
signal.signal(signal.SIGTERM, signal_handler)  # 終止信號(hào)

4-2：方法二

import atexit
from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler(timezone="Asia/Shanghai")
scheduler.add_job(sync_analysis_s7_data, "interval", seconds=10)
scheduler.start()

def cleanup():
    """清理函數(shù)"""
    print("正在清理資源...")
    scheduler.shutdown()

# 注冊(cè)退出時(shí)的清理函數(shù)
atexit.register(cleanup)

4-3：方法三

只是依賴FastApi框架

@app.on_event("shutdown")
async def shutdown_event():
    """應(yīng)用關(guān)閉時(shí)執(zhí)行"""
    # 停止定時(shí)任務(wù)
    scheduler.shutdown()
    logger.info("===== AnalysisS7數(shù)據(jù)服務(wù)關(guān)閉 =====")