用于实时日志记录的 Python tkinter 文本小部件

Question

我正在尝试为我的 python 应用程序创建一个 GUI，我正在努力让 Text Widget 表现得像一个标准输出。更具体地说，我希望小部件在请求时立即显示信息，而不是在一切完成后显示。

为了演示，考虑一下：

import sys
import time
import multiprocessing as mp
import multiprocessing.queues as mpq
from threading import Thread
from tkinter import *


class StdoutQueue(mpq.Queue):

    def __init__(self, *args, **kwargs):
        ctx = mp.get_context()
        super(StdoutQueue, self).__init__(*args, **kwargs, ctx=ctx)

    def write(self,msg):
        self.put(msg)

    def flush(self):
        sys.__stdout__.flush()
        
        
def text_catcher(text_widget,queue):
    while True:
        text_widget.insert(END, queue.get())


def test_child(q):
    sys.stdout = q
    print('child running')


def test_parent(q):
    sys.stdout = q
    print('parent running')
    time.sleep(5.)
    mp.Process(target=test_child,args=(q,)).start()


if __name__ == '__main__':
    gui_root = Tk()
    gui_txt = Text(gui_root)
    gui_txt.pack()
    q = StdoutQueue()
    gui_btn = Button(gui_root, text='Test', command=lambda: test_parent(q),)
    gui_btn.pack()

    monitor = Thread(target=text_catcher, args=(gui_txt, q))
    monitor.daemon = True
    monitor.start()

    gui_root.mainloop()

点击Test 按钮不会立即打印'parent running' 并在5 秒后打印'child running'，而是等待5 秒并同时打印两个字符串。

有什么方法可以得到想要的行为？

Answer 1

我完全不明白发生了什么。当从 Tk 线程调用 test_parent 时，text_catcher 被阻塞。我会假设监视器线程可以防止这种情况。嗯……

一种可能的解决方法是将test_parent 放在它自己的线程中。这可以通过一个小装饰器轻松完成：

def run_in_thread(fun):
    def wrapper(*args, **kwargs):
        thread = Thread(target=fun, args=args, kwargs=kwargs)
        thread.start()
        return thread

    return wrapper


@run_in_thread
def test_parent(q):
    sys.stdout = q
    print('parent running')
    time.sleep(5.)
    mp.Process(target=test_child, args=(q,)).start()

test_parent 像这样包装，短信立即出现。

Answer 2

只需将 time.sleep(5.) 从 test_parent 移动到 test_child 中：

def test_child(q):
    sys.stdout = q
    time.sleep(5.)
    print('child running')


def test_parent(q):
    sys.stdout = q
    print('parent running')
    mp.Process(target=test_child, args=(q,)).start()

即使我在没有StdoutQueue 的情况下尝试它也可以正常工作，这里是代码：

import time
import multiprocessing as mp
from tkinter import *


def test_child():
    time.sleep(5.)
    print('child running')


def test_parent():
    print('parent running')
    mp.Process(target=test_child, args=()).start()


if __name__ == '__main__':
    gui_root = Tk()
    gui_txt = Text(gui_root)
    gui_txt.pack()
    gui_btn = Button(gui_root, text='Test', command=lambda: test_parent(), )
    gui_btn.pack()
    gui_root.mainloop()