如何通过APScheduler获取调度任务关联的系统进程详情？

好问题！APScheduler本身并没有内置的函数直接返回任务关联的进程信息——它的核心定位是任务调度器，而非进程管理工具。不过别担心，我们可以通过手动跟踪子进程+辅助库的方式实现你的需求，毕竟当你用APScheduler调度外部命令（比如pinparsing）时，本质是在Python中启动了一个子进程，只要我们把这个进程的引用保存下来，就能随时查询它的详情。

核心思路 #

1. 用subprocess.Popen启动外部命令时，保留进程对象的引用
2. 将进程对象与APScheduler的任务ID做映射存储（比如用字典）
3. 结合get_jobs()获取任务列表，再通过任务ID匹配对应的进程对象
4. 借助psutil库获取进程的详细信息（PID、资源占用、状态等）

实现代码 #

首先安装依赖：

pip install apscheduler psutil

以下是完整的示例代码：

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

# 全局字典：任务ID -> 进程对象，用来关联任务和进程
job_process_map = {}

def run_external_command():
    """执行外部命令并保存进程引用"""
    # 替换成你的pinparsing命令，可根据需求添加参数
    proc = subprocess.Popen(
        ["pinparsing"],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True
    )
    
    # 获取当前任务的ID，存入映射字典
    current_job = BackgroundScheduler.get_current_job()
    if current_job:
        job_process_map[current_job.id] = proc
        print(f"任务 {current_job.id} 关联进程PID: {proc.pid}")
    
    return proc

if __name__ == "__main__":
    # 初始化后台调度器
    scheduler = BackgroundScheduler()
    
    # 添加延后执行的任务（这里用date触发器，立即执行可替换为datetime.now()）
    job = scheduler.add_job(
        run_external_command,
        "date",
        run_date=datetime.now()
    )
    print(f"已添加任务，ID: {job.id}")
    
    # 启动调度器
    scheduler.start()
    
    # 等待任务启动执行
    time.sleep(2)
    
    # 查询任务及关联进程信息
    print("\n===== 任务与进程详情 =====")
    for job in scheduler.get_jobs():
        print(f"\n任务ID: {job.id}")
        print(f"任务状态: {job.state}")
        
        if job.id in job_process_map:
            proc = job_process_map[job.id]
            # 基础进程信息
            print(f"进程PID: {proc.pid}")
            print(f"进程状态: {'运行中' if proc.poll() is None else f'已结束，退出码: {proc.poll()}'}")
            
            # 用psutil获取更详细的系统级信息
            try:
                ps_proc = psutil.Process(proc.pid)
                print(f"进程名称: {ps_proc.name()}")
                print(f"CPU占用率: {ps_proc.cpu_percent(interval=0.5)}%")
                print(f"内存占用: {ps_proc.memory_info().rss / 1024 / 1024:.2f} MB")
                print(f"进程创建时间: {datetime.fromtimestamp(ps_proc.create_time()).strftime('%Y-%m-%d %H:%M:%S')}")
            except psutil.NoSuchProcess:
                print("⚠️ 该进程已终止或不存在")
        else:
            print("⚠️ 未找到关联的进程信息（任务可能未执行或执行的是Python函数而非外部命令）")
    
    # 优雅关闭调度器和进程（实际应用中可根据需求调整）
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n正在关闭调度器...")
        scheduler.shutdown()
        # 终止所有运行中的关联进程
        for proc in job_process_map.values():
            if proc.poll() is None:
                proc.terminate()
                proc.wait()
        print("已完成清理")

关键注意事项 #

• 如果你的任务是执行Python函数（而非外部命令），那么任务会在调度器所在的Python进程中运行，此时获取的进程信息就是当前Python进程的信息。
• job_process_map需要是全局或任务函数可访问的对象，确保任务执行时能正确存储进程引用。
• 记得定期清理job_process_map中已结束的进程条目，避免内存冗余。
• 不同类型的APScheduler调度器（如BlockingScheduler）代码逻辑类似，只需调整启动方式即可。

内容的提问来源于stack exchange，提问作者Exprator