当前位置：首页 > news >正文

实时输出subprocess.Popen运行程序的日志

news 2025/8/6 21:58:46

实时输出subprocess.Popen运行程序的日志，因为readline会阻塞，所以开启两个线程分别接受标准输出和标准错误

import logging
import os
import subprocess
import sys
import threading
import timelogging.basicConfig(level=logging.INFO)class ProcessLogRecorder(threading.Thread):def __init__(self, proc, is_stderr=False, log_name="", sleep_secs=0.2):super().__init__()self._proc = procself._is_stderr = is_stderrself._stream = self._proc.stderr if is_stderr else self._proc.stdoutself._log = logging.getLogger(log_name)self._sleep_secs = sleep_secsself._running = threading.Event()self._running.set()def work(self):while self._running.is_set() and self._proc.poll() is None:line = self._stream.readline()if not line:time.sleep(self._sleep_secs)if self._is_stderr:self._log.error(line)else:self._log.info(line)def stop(self):self._running.clear()def run(self):try:self.work()except Exception as ex:print(ex)class ManageSingleServer:@staticmethoddef start_server():cmd = ['ping', 'www.baidu.com']cmd = ["timeout", "/t", "5"]proc = subprocess.Popen(cmd,shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)stdout_recorder = ProcessLogRecorder(proc)stderr_recorder = ProcessLogRecorder(proc, True)stdout_recorder.start()stderr_recorder.start()return proc, stdout_recorder, stderr_recorder@classmethoddef run(cls):proc, stdout_recorder, stderr_recorder = cls.start_server()try:for i in range(20):print(proc.poll(), stdout_recorder.is_alive())time.sleep(2)except KeyboardInterrupt:print("Ctrl+C pressed")proc.terminate()proc.wait()stdout_recorder.stop()stderr_recorder.stop()sys.exit(0)if __name__ == "__main__":ManageSingleServer.run()