如何自定义调度运行同一脚本两个实例并通过stdin/stdout传递参数与输出

优先选择 subprocess 实现需求，另外两种方案均不匹配场景：

• 排除线程：线程属于同一进程空间，无法独立执行外部的connector.py脚本，也无法为每个实例分配独立的stdin/stdout文件描述符，完全不符合需求。
• 排除multiprocessing.Process：该类设计用于启动同项目内的Python函数作为子进程，若要执行独立的connector.py脚本需要额外封装，还要手动处理IO通道重定向，复杂度远高于subprocess。
• 选择subprocess：原生支持启动独立的Python脚本进程，直接支持对stdin/stdout/stderr的通道重定向，完全匹配你要通过stdin传配置、stdout收输出的要求，每个子进程的IO通道天然隔离，不会串数据。

每个connector.py实例启动时单独创建子进程，各自维护独立的IO通道，不会互相干扰：

1. 启动子进程时配置IO参数：用subprocess.Popen启动，指定stdin=subprocess.PIPE、stdout=subprocess.PIPE、text=True（兼容旧版本Python可换为universal_newlines=True），可直接读写字符串格式的配置和输出。
2. 传递配置：启动子进程后，通过子进程对象.communicate(input=配置序列化字符串)把对应实例的配置写入stdin，不需要用命令行参数，完全符合要求。结构化配置可以序列化为JSON字符串传递，connector.py侧直接从stdin读全量数据反序列化即可。
3. 接收输出：communicate方法的返回值直接包含子进程的stdout输出内容，不会和其他实例的输出混淆。
4. 资源清理：communicate会自动等待子进程执行结束，后续可调用wait()确保进程资源回收，避免出现僵尸进程。

无需第三方调度库，极简实现逻辑如下：

• 提前为两个connector实例定义各自的配置、运行间隔、上次运行时间、运行状态标记
• 主循环用time.sleep(1)做低频率轮询，避免空转占满CPU
• 每次循环检查当前时间和两个实例上次运行时间的差值，达到间隔阈值且上一次任务已执行完成的情况下，启动对应实例的子进程，执行IO交互逻辑，更新上次运行时间

调度器侧代码 #

import subprocess
import time
import json

# 两个实例的独立配置
INSTANCE_CONFIGS = [
    {"id": 1, "interval": 60, "api_key": "xxx", "pull_target": "a", "last_run": 0, "running": False},
    {"id": 2, "interval": 300, "api_key": "yyy", "pull_target": "b", "last_run": 0, "running": False}
]

def run_connector(config):
    config["running"] = True
    # 启动子进程
    proc = subprocess.Popen(
        ["python", "connector.py"],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True
    )
    # 通过stdin传递配置，同时获取stdout输出
    stdout, stderr = proc.communicate(input=json.dumps(config))
    # 按需处理输出和错误日志
    print(f"实例{config['id']}输出：{stdout}")
    if stderr:
        print(f"实例{config['id']}错误：{stderr}")
    proc.wait()
    config["running"] = False
    config["last_run"] = time.time()

if __name__ == "__main__":
    while True:
        now = time.time()
        for conf in INSTANCE_CONFIGS:
            if not conf["running"] and now - conf["last_run"] >= conf["interval"]:
                run_connector(conf)
        time.sleep(1)

connector.py侧参考逻辑 #

import sys
import json

if __name__ == "__main__":
    # 从stdin读取全量配置
    config = json.loads(sys.stdin.read())
    # 此处写你的API拉取、结果写文件逻辑
    # 最终输出到stdout回传给调度器
    print(json.dumps({"status": "success", "task_id": config["id"]}))

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