如何将Python subprocess句柄从Module_B传递给Module_C以控制Module_A？

直接传递Module_B中的proc句柄不可行——因为Module_B退出后，其进程内的对象会被系统回收，无法跨进程传递给Module_C。以下是几种可行的替代方案：

方案1：通过进程ID（PID）+ 独立通信通道 #

1. Module_B启动Module_A的子进程后，将PID写入一个本地文件（比如module_a_pid.txt）
2. Module_C读取PID文件确认Module_A存活，然后通过本地Socket/命名管道与Module_A通信
3. Module_A需要提前实现对应的监听逻辑，接收并处理命令

示例代码片段：

• Module_B.py：

import subprocess

proc = subprocess.Popen(["python", "Module_A.py"])
with open("module_a_pid.txt", "w") as f:
    f.write(str(proc.pid))

• Module_A.py（添加Unix Socket监听）：

import socket
import threading

def command_listener():
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    sock.bind("/tmp/module_a_cmd.sock")
    sock.listen(1)
    global running
    while running:
        conn, _ = sock.accept()
        cmd = conn.recv(1024).decode().strip()
        if cmd == "terminate":
            running = False
            break
        # 处理其他命令
        conn.close()

running = True
# 启动后台监听线程
threading.Thread(target=command_listener, daemon=True).start()

# 原有业务逻辑
while running:
    # 从队列/管道获取输入、执行函数
    pass

• Module_C.py：

import socket

sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.connect("/tmp/module_a_cmd.sock")
sock.send(b"terminate")
sock.close()

方案2：使用命名管道直接通信 #

跳过PID传递，提前创建命名管道作为Module_A和Module_C的通信桥梁：

1. Module_A启动后监听命名管道的输入
2. Module_C直接向管道写入命令，无需依赖Module_B的任何资源

示例（Linux/macOS）：

• Module_A.py：

import os

PIPE_PATH = "/tmp/module_a_pipe"
if not os.path.exists(PIPE_PATH):
    os.mkfifo(PIPE_PATH)

# 监听管道命令
with open(PIPE_PATH, "r") as pipe:
    while True:
        cmd = pipe.readline().strip()
        if cmd == "terminate":
            break
        # 处理其他命令

• Module_C.py：

PIPE_PATH = "/tmp/module_a_pipe"
with open(PIPE_PATH, "w") as pipe:
    pipe.write("terminate\n")

方案3：使用中间件做消息中转 #

适合复杂通信场景，用Redis、MQ等中间件作为命令传递的中介：

1. Module_A启动后订阅中间件的指定频道/队列
2. Module_C向对应频道/队列发送命令，Module_A接收并处理

示例（Redis）：

• Module_A.py：

import redis

r = redis.Redis()
pubsub = r.pubsub()
pubsub.subscribe("module_a_commands")

running = True
while running:
    msg = pubsub.get_message()
    if msg and msg["type"] == "message":
        cmd = msg["data"].decode()
        if cmd == "terminate":
            running = False
            break
        # 处理其他命令
    # 原有业务逻辑

• Module_C.py：

import redis

r = redis.Redis()
r.publish("module_a_commands", "terminate")

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