关于Python调用(创建)其他进程

目录

1. Python 启动子进程并实时输出日志的完整指南
2. 核心实现解析
3. 1. 实时日志输出机制
4. 2. 启动配置
5. 3. 线程管理
6. 使用示例
7. 示例 1：基本使用
8. 示例 2：复杂命令行
9. 示例 3：长期运行的服务
10. 高级用法
11. 1. 日志过滤
12. 2. 日志转发到文件
13. 3. 进度监控
14. 4. 多进程管理
15. 常见问题解决方案

Python 启动子进程并实时输出日志的完整指南#

下面是一个完整的解决方案，用于启动子进程并实时将其日志输出到终端：

import subprocess
import sys
import threading
import time
import os
import signal
import psutil

class ProcessManager:
    def __init__(self, executable_path, *args):
        """
        进程管理器
        :param executable_path: 可执行文件路径
        :param args: 命令行参数
        """
        self.executable_path = executable_path
        self.args = list(args)
        self.process = None
        self.pid = None
        self.stdout_thread = None
        self.stderr_thread = None
        self.stdout_lines = []
        self.stderr_lines = []
        self.running = False

        # Windows特定配置
        self.startupinfo = None
        if sys.platform == "win32":
            self.startupinfo = subprocess.STARTUPINFO()
            self.startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
            self.startupinfo.wShowWindow = subprocess.SW_HIDE

    def start(self):
        """启动进程并实时输出日志"""
        if self.is_running():
            print(f"进程已在运行 (PID: {self.pid})")
            return False

        try:
            # 创建进程
            self.process = subprocess.Popen(
                [self.executable_path] + self.args,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                bufsize=1,  # 行缓冲
                text=True,  # 文本模式
                startupinfo=self.startupinfo
            )
            self.pid = self.process.pid
            self.running = True

            print(f"进程已启动 (PID: {self.pid}): {self.executable_path} {' '.join(self.args)}")

            # 启动线程实时输出日志
            self.stdout_thread = threading.Thread(
                target=self._read_stream,
                args=(self.process.stdout, "STDOUT")
            )
            self.stderr_thread = threading.Thread(
                target=self._read_stream,
                args=(self.process.stderr, "STDERR")
            )

            self.stdout_thread.daemon = True
            self.stderr_thread.daemon = True

            self.stdout_thread.start()
            self.stderr_thread.start()

            return True
        except FileNotFoundError:
            print(f"错误: 找不到可执行文件: {self.executable_path}")
            return False
        except PermissionError:
            print(f"错误: 没有执行权限: {self.executable_path}")
            return False

    def _read_stream(self, stream, stream_name):
        """从流中读取数据并实时输出"""
        try:
            for line in iter(stream.readline, ''):
                if not line:
                    break
                line = line.rstrip()
                # 实时输出到终端
                print(f"[{stream_name} PID:{self.pid}] {line}")
                # 保存日志
                if stream_name == "STDOUT":
                    self.stdout_lines.append(line)
                else:
                    self.stderr_lines.append(line)
        except ValueError:
            # 当流关闭时可能发生
            pass
        finally:
            stream.close()

    def is_running(self):
        """检查进程是否正在运行"""
        return self.running and self.process and (self.process.poll() is None)

    def wait(self):
        """等待进程结束"""
        if self.process:
            return self.process.wait()
        return None

    def stop(self, timeout=5):
        """停止进程"""
        if not self.is_running():
            print("进程未运行")
            return True

        print(f"正在停止进程 (PID: {self.pid})...")

        # 优雅终止
        try:
            if sys.platform == "win32":
                self.process.terminate()
            else:
                os.kill(self.pid, signal.SIGTERM)

            # 等待结束
            start_time = time.time()
            while time.time() - start_time < timeout:
                if not self.is_running():
                    print("进程已终止")
                    return True
                time.sleep(0.1)
        except Exception as e:
            print(f"优雅终止失败: {str(e)}")

        # 强制终止
        try:
            self._force_kill()
            print("进程已强制终止")
            return True
        except Exception as e:
            print(f"强制终止失败: {str(e)}")
            return False

    def _force_kill(self):
        """强制终止进程树"""
        try:
            parent = psutil.Process(self.pid)
            children = parent.children(recursive=True)

            # 终止子进程
            for child in children:
                try:
                    child.kill()
                except psutil.NoSuchProcess:
                    pass

            # 终止父进程
            parent.kill()

            # 等待结束
            parent.wait(timeout=2)
        except psutil.NoSuchProcess:
            pass
        except Exception:
            # 回退
            self.process.kill()
            self.process.wait()
        finally:
            self.running = False

    def get_logs(self, stream="both", max_lines=None):
        """获取日志"""
        result = {}

        if stream in ("both", "stdout"):
            lines = self.stdout_lines
            if max_lines and len(lines) > max_lines:
                lines = lines[-max_lines:]
            result["stdout"] = "\\n".join(lines)

        if stream in ("both", "stderr"):
            lines = self.stderr_lines
            if max_lines and len(lines) > max_lines:
                lines = lines[-max_lines:]
            result["stderr"] = "\\n".join(lines)

        return result

    def __del__(self):
        """确保进程被终止"""
        if self.is_running():
            self.stop()

def main():
    # 创建进程管理器
    process_manager = ProcessManager("xxxx.exe", "-f", "app.yaml")

    # 启动进程并实时输出日志
    if not process_manager.start():
        print("启动失败")
        return

    try:
        # 主线程可以继续工作
        print("主线程正在工作...")

        # 模拟工作
        for i in range(10):
            print(f"主线程工作状态: {i+1}/10")
            time.sleep(1)

            # 检查进程状态
            if not process_manager.is_running():
                print("子进程已提前退出")
                break

        print("主线程工作完成")
    finally:
        # 确保停止进程
        process_manager.stop()

        # 获取并保存日志
        logs = process_manager.get_logs()
        with open("process_logs.txt", "w") as f:
            f.write("=== STDOUT ===\\n")
            f.write(logs.get("stdout", "") + "\\n\\n")
            f.write("=== STDERR ===\\n")
            f.write(logs.get("stderr", "") + "\\n")

        print("日志已保存到 process_logs.txt")

if __name__ == "__main__":
    main()

核心实现解析#

1. 实时日志输出机制#

def _read_stream(self, stream, stream_name):
    """从流中读取数据并实时输出"""
    try:
        for line in iter(stream.readline, ''):
            if not line:
                break
            line = line.rstrip()
            # 实时输出到终端
            print(f"[{stream_name} PID:{self.pid}] {line}")
            # 保存日志
            if stream_name == "STDOUT":
                self.stdout_lines.append(line)
            else:
                self.stderr_lines.append(line)
    finally:
        stream.close()

关键点：

• 使用 iter(stream.readline, '') 循环读取每一行
• 使用 print() 实时输出到终端
• 同时保存日志到内存列表
• 使用独立的线程处理 stdout 和 stderr

2. 启动配置#

self.process = subprocess.Popen(
    [self.executable_path] + self.args,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    bufsize=1,  # 行缓冲 - 确保实时输出
    text=True,  # 文本模式 - 避免处理字节
    startupinfo=self.startupinfo  # Windows 隐藏窗口
)

关键参数：

• bufsize=1：启用行缓冲，确保实时输出
• text=True：以文本模式处理输出，避免处理字节数据
• startupinfo：Windows 下隐藏子进程窗口

3. 线程管理#

# 启动线程实时输出日志
self.stdout_thread = threading.Thread(
    target=self._read_stream,
    args=(self.process.stdout, "STDOUT")
)
self.stderr_thread = threading.Thread(
    target=self._read_stream,
    args=(self.process.stderr, "STDERR")
)

self.stdout_thread.daemon = True
self.stderr_thread.daemon = True

self.stdout_thread.start()
self.stderr_thread.start()

特点：

• 使用守护线程 (daemon=True)，确保主线程退出时自动结束
• 独立线程处理 stdout 和 stderr
• 线程安全地收集日志

使用示例#

示例 1：基本使用#

manager = ProcessManager("python", "-c", "import time; print('开始'); time.sleep(5); print('结束')")
manager.start()

# 主线程可以继续工作
time.sleep(2)
print("主线程工作中...")

manager.wait()

示例 2：复杂命令行#

# 启动带复杂参数的进程
ffmpeg = ProcessManager(
    "ffmpeg",
    "-i", "input.mp4",
    "-c:v", "libx264",
    "-crf", "23",
    "-preset", "medium",
    "-c:a", "aac",
    "-b:a", "192k",
    "output.mp4"
)

if ffmpeg.start():
    try:
        # 实时查看转换进度
        while ffmpeg.is_running():
            time.sleep(1)
            print("转换中...")
    finally:
        ffmpeg.stop()

示例 3：长期运行的服务#

# 启动Web服务器
server = ProcessManager("python", "-m", "http.server", "8080")

server.start()

try:
    print("服务器运行中，按Ctrl+C停止...")
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    print("停止服务器...")
    server.stop()

高级用法#

1. 日志过滤#

def _read_stream(self, stream, stream_name):
    for line in iter(stream.readline, ''):
        line = line.rstrip()

        # 过滤特定日志
        if "ERROR" in line:
            print(f"\\033[91m[{stream_name}] {line}\\033[0m")  # 红色显示错误
        elif "WARN" in line:
            print(f"\\033[93m[{stream_name}] {line}\\033[0m")  # 黄色显示警告
        else:
            print(f"[{stream_name}] {line}")

        # 保存日志...

2. 日志转发到文件#

def start(self):
    # ...
    self.log_file = open("application.log", "a")
    # ...

def _read_stream(self, stream, stream_name):
    for line in iter(stream.readline, ''):
        line = line.rstrip()
        print(f"[{stream_name}] {line}")

        # 同时写入文件
        self.log_file.write(f"[{time.ctime()}] [{stream_name}] {line}\\n")
        self.log_file.flush()

        # 保存日志...

3. 进度监控#

def _read_stream(self, stream, stream_name):
    for line in iter(stream.readline, ''):
        line = line.rstrip()
        print(f"[{stream_name}] {line}")

        # 解析进度信息
        if "Progress:" in line:
            try:
                percent = int(line.split(":")[1].strip().rstrip("%"))
                self.update_progress(percent)
            except:
                pass

        # 保存日志...

4. 多进程管理#

class ProcessPool:
    def __init__(self):
        self.processes = []

    def add_process(self, executable, *args):
        manager = ProcessManager(executable, *args)
        self.processes.append(manager)
        return manager

    def start_all(self):
        for proc in self.processes:
            proc.start()

    def stop_all(self):
        for proc in self.processes:
            proc.stop()

    def monitor(self):
        while any(proc.is_running() for proc in self.processes):
            time.sleep(1)
            # 显示所有进程状态...

# 使用
pool = ProcessPool()
pool.add_process("worker1.exe", "-c", "config1.yaml")
pool.add_process("worker2.exe", "-c", "config2.yaml")
pool.start_all()
pool.monitor()

常见问题解决方案#

Q: 日志输出有延迟？
A: 确保设置了 bufsize=1 和 text=True，使用行缓冲模式 Q: Windows下子进程窗口仍然出现？
A: 确保正确设置了 startupinfo 并且程序是控制台应用 Q: 输出混乱或交错？
A: 使用锁确保线程安全：

from threading import Lock
print_lock = Lock()

def _read_stream(self, stream, stream_name):
    for line in iter(stream.readline, ''):
        with self.print_lock:
            print(f"[{stream_name}] {line.rstrip()}")
        # ...

Q: 子进程卡住导致主线程无法退出？
A: 使用带超时的等待：

def wait(self, timeout=None):
    try:
        return self.process.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        return None

Q: 如何同时输出到终端和文件？
A: 使用 tee 命令（Unix）或 Python 双重输出：

# 在 _read_stream 方法中
print(f"[{stream_name}] {line}")
self.log_file.write(f"{line}\\n")
self.log_file.flush()

这个方案提供了完整的进程管理功能，特别是实时日志输出到终端的能力，适用于各种需要监控子进程输出的场景。