假设您正在调用的进程是长时间运行的并且不会一次性产生所有输出，这意味着您不能使用subprocess.Popen.communicate()，因为它旨在读取所有输出直到结束文件。
您将不得不使用其他标准技术从管道中读取数据。
由于您希望将其与 GUI 集成并且该过程需要长时间运行，因此您需要协调读取其输出与 GUI 的主循环。这使事情变得有些复杂。
TkInter
让我们首先假设您想使用 TkInter，就像在您的一个示例中一样。这给我们带来了几个问题：
- TkInter 没有与 select 模块集成。
- 到目前为止，甚至还没有 TkInter 与 asyncio 的规范集成（另请参阅 https://bugs.python.org/issue27546）。
- 通常建议不要使用
root.update() 将自定义主循环组合在一起，这样我们就可以通过线程来解决本应基于事件的方法。
- TkInter 的
event_generate() 缺少 Tk 将用户数据与事件一起发送的能力，因此我们不能使用 TkInter 事件将接收到的输出从一个线程传递到另一个线程。
因此，我们将使用 threading（即使我不想这样做）来解决它，其中主线程控制 Tk GUI，辅助线程读取进程的输出，并且缺少作为 TkInter 中传递数据的原生方式，我们使用线程安全的 Queue。
#!/usr/bin/env python3
from subprocess import Popen, PIPE, STDOUT, TimeoutExpired
from threading import Thread, Event
from queue import Queue, Empty
from tkinter import Tk, Text, END
class ProcessOutputReader(Thread):
def __init__(self, queue, cmd, params=(),
group=None, name=None, daemon=True):
super().__init__(group=group, name=name, daemon=daemon)
self._stop_request = Event()
self.queue = queue
self.process = Popen((cmd,) + tuple(params),
stdout=PIPE,
stderr=STDOUT,
universal_newlines=True)
def run(self):
for line in self.process.stdout:
if self._stop_request.is_set():
# if stopping was requested, terminate the process and bail out
self.process.terminate()
break
self.queue.put(line) # enqueue the line for further processing
try:
# give process a chance to exit gracefully
self.process.wait(timeout=3)
except TimeoutExpired:
# otherwise try to terminate it forcefully
self.process.kill()
def stop(self):
# request the thread to exit gracefully during its next loop iteration
self._stop_request.set()
# empty the queue, so the thread will be woken up
# if it is blocking on a full queue
while True:
try:
self.queue.get(block=False)
except Empty:
break
self.queue.task_done() # acknowledge line has been processed
class MyConsole(Text):
def __init__(self, parent, queue, update_interval=50, process_lines=500):
super().__init__(parent)
self.queue = queue
self.update_interval = update_interval
self.process_lines = process_lines
self.after(self.update_interval, self.fetch_lines)
def fetch_lines(self):
something_inserted = False
for _ in range(self.process_lines):
try:
line = self.queue.get(block=False)
except Empty:
break
self.insert(END, line)
self.queue.task_done() # acknowledge line has been processed
# ensure scrolling the view is at most done once per interval
something_inserted = True
if something_inserted:
self.see(END)
self.after(self.update_interval, self.fetch_lines)
# create the root widget
root = Tk()
# create a queue for sending the lines from the process output reader thread
# to the TkInter main thread
line_queue = Queue(maxsize=1000)
# create a process output reader
reader = ProcessOutputReader(line_queue, 'python3', params=['-u', 'test.py'])
# create a console
console = MyConsole(root, line_queue)
reader.start() # start the process
console.pack() # make the console visible
root.mainloop() # run the TkInter main loop
reader.stop()
reader.join(timeout=5) # give thread a chance to exit gracefully
if reader.is_alive():
raise RuntimeError("process output reader failed to stop")
由于前面提到的警告，TkInter 代码最终有点偏大。
PyQt
使用 PyQt 代替，我们可以大大改善我们的情况，因为该框架已经提供了一种与子流程集成的本地方式，其形式为 QProcess 类.
这意味着我们可以取消线程并改用 Qt 的原生 Signal 和 Slot 机制。
#!/usr/bin/env python3
import sys
from PyQt5.QtCore import pyqtSignal, pyqtSlot, QProcess, QTextCodec
from PyQt5.QtGui import QTextCursor
from PyQt5.QtWidgets import QApplication, QPlainTextEdit
class ProcessOutputReader(QProcess):
produce_output = pyqtSignal(str)
def __init__(self, parent=None):
super().__init__(parent=parent)
# merge stderr channel into stdout channel
self.setProcessChannelMode(QProcess.MergedChannels)
# prepare decoding process' output to Unicode
codec = QTextCodec.codecForLocale()
self._decoder_stdout = codec.makeDecoder()
# only necessary when stderr channel isn't merged into stdout:
# self._decoder_stderr = codec.makeDecoder()
self.readyReadStandardOutput.connect(self._ready_read_standard_output)
# only necessary when stderr channel isn't merged into stdout:
# self.readyReadStandardError.connect(self._ready_read_standard_error)
@pyqtSlot()
def _ready_read_standard_output(self):
raw_bytes = self.readAllStandardOutput()
text = self._decoder_stdout.toUnicode(raw_bytes)
self.produce_output.emit(text)
# only necessary when stderr channel isn't merged into stdout:
# @pyqtSlot()
# def _ready_read_standard_error(self):
# raw_bytes = self.readAllStandardError()
# text = self._decoder_stderr.toUnicode(raw_bytes)
# self.produce_output.emit(text)
class MyConsole(QPlainTextEdit):
def __init__(self, parent=None):
super().__init__(parent=parent)
self.setReadOnly(True)
self.setMaximumBlockCount(10000) # limit console to 10000 lines
self._cursor_output = self.textCursor()
@pyqtSlot(str)
def append_output(self, text):
self._cursor_output.insertText(text)
self.scroll_to_last_line()
def scroll_to_last_line(self):
cursor = self.textCursor()
cursor.movePosition(QTextCursor.End)
cursor.movePosition(QTextCursor.Up if cursor.atBlockStart() else
QTextCursor.StartOfLine)
self.setTextCursor(cursor)
# create the application instance
app = QApplication(sys.argv)
# create a process output reader
reader = ProcessOutputReader()
# create a console and connect the process output reader to it
console = MyConsole()
reader.produce_output.connect(console.append_output)
reader.start('python3', ['-u', 'test.py']) # start the process
console.show() # make the console visible
app.exec_() # run the PyQt main loop
我们最终得到了一个源自 Qt 类的小样板，但采用了一种整体更简洁的方法。
一般注意事项
还要确保您正在调用的进程没有缓冲多个输出行，否则它仍然看起来好像控制台卡住了。
特别是如果被调用者是 python 程序，您可以确保它使用 print(..., flush=True) 或使用 python -u callee.py 调用它以强制执行无缓冲输出。