如何正确在Python中使用subprocess.Popen与线程？代码执行异常求助

我正在编写Python脚本，需通过两个bash终端依次执行其他脚本，并将输出信息写入Tkinter文本框。已使用subprocess.Popen创建两个bash进程，通过线程监控每个进程的stdout和stderr，同时用while循环控制执行流程。但点击Start按钮后仅输出Button clicked，while循环未按预期依次执行命令。

原脚本代码：

import os
import tkinter as tk
from tkinter import *
import tkinter.scrolledtext as tkst
import subprocess
from subprocess import Popen
import threading

class TKValidator:
    def __init__(self): 
        #create TK
        self.ws = Tk()
        self.ws.state('zoomed')
        
        self.flag = IntVar()
        
        #process for bash 1
        self.process1 = subprocess.Popen(["bash"], stderr=subprocess.PIPE,shell=False, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
        #process for bash 2
        self.process2 = subprocess.Popen(["bash"], stderr=subprocess.PIPE,shell=False, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
        exit = False
        
        #start threads (stdout and stderror monitor)
        threading.Thread(target=self.read_stdout_1).start()
        threading.Thread(target=self.read_stderror_1).start()
        threading.Thread(target=self.read_stdout_2).start()
        threading.Thread(target=self.read_stderror_2).start()
        
        self.frame2 = tk.Frame(self.ws, bg="#777474")
        self.frame2.grid(column=0, row=0, padx=(100,100), pady=(10, 10), sticky="W")
        
        #Start button
        self.btnVerify = Button(self.ws, text='Start', command=self.ButtonClick)
        self.btnVerify.grid(row=5, columnspan=3, pady=10)
        
        #output text area
        self.logArea = tkst.ScrolledText(self.ws, wrap= tk.WORD, width=80, height=20, name="logArea")
        self.logArea.grid(padx=(100,100), pady=(10, 10) ,row=10, sticky="W")


    def read_stdout_1(self):
        while not exit:
            msg = self.process1.stdout.readline()
            self.logArea.insert(INSERT, "process_1 logging start")
            self.logArea.insert(INSERT, msg.decode())
            self.logArea.insert(END, "\n logging end")
            
    def read_stderror_1(self):
        while not exit:
            msg = self.process1.stderr.readline()
            self.logArea.insert(INSERT, "Error")
            self.logArea.insert(INSERT, msg.decode())
            self.logArea.insert(END, "\n Error end")


    def read_stdout_2(self):
        while not exit:
            msg = self.process2.stdout.readline()
            self.logArea.insert(INSERT, "process_2 logging start")
            self.logArea.insert(INSERT, msg.decode())
            self.logArea.insert(END, "\logging end")

            
    def read_stderror_2(self):
        while not exit:
            msg = self.process2.stderr.readline()
            self.logArea.insert(INSERT, "Error")
            self.logArea.insert(INSERT, msg.decode())
            self.logArea.insert(END, "\n Error end")


    def ButtonClick(self):
        print("Button clicked")
        
        #set flag to 1
        self.flag = 1
        
        while not exit:
            if self.flag == 1:
                print("Case 1")
                self.process1.stdin.write(r'python C:\path\test1.py'.encode())
                self.process1.stdin.flush()
                self.flag = 2
                
            elif self.flag == 2:
                print("Case 2")
                self.process1.stdin.write(r'python C:\path\test2.py'.encode())
                self.process1.stdin.flush()
                self.flag = 3
            
            elif self.flag == 3:
                print("Case 3")
                self.process2.stdin.write(r'python C:\path\test3.py'.encode())
                self.process2.stdin.flush()
                self.flag = 4
           
            elif self.flag == 4:
                print("Case 3")
                self.process2.stdin.write(r'python C:\path\test4.py'.encode())
                self.process2.stdin.flush()
                self.flag = 5
                
            else:
                break


    def run(self):
        self.ws.title('Test Script')
        self.ws.mainloop() 
        
validator = TKValidator()
validator.run() 

问题分析 #

1. 变量作用域错误：__init__中定义的exit是局部变量，线程函数和ButtonClick中的while not exit无法访问该变量，会触发NameError，导致线程直接终止，ButtonClick的循环也无法正常执行。
2. IntVar使用错误：初始化时self.flag = IntVar()是Tkinter的变量对象，但ButtonClick中直接赋值self.flag = 1将其覆盖为整数，破坏了Tkinter变量的特性，后续判断逻辑混乱。
3. bash命令未结束：向bash进程写入命令时未添加换行符\n，bash会一直等待输入结束，不会执行命令。
4. GUI线程不安全操作：线程中直接调用self.logArea.insert会导致Tkinter主线程与子线程的GUI操作冲突，可能引发界面假死或异常。
5. 主线程被阻塞：ButtonClick在Tkinter主线程中执行，里面的while循环会持续占用主线程，导致GUI无法响应，后续步骤无法触发。

修复方案 #

以下是修复后的完整代码：

import tkinter as tk
import tkinter.scrolledtext as tkst
import subprocess
import threading

class TKValidator:
    def __init__(self): 
        self.ws = Tk()
        self.ws.state('zoomed')
        
        # 使用普通整数变量替代IntVar，简化逻辑
        self.flag = 0
        # 实例化exit变量，确保所有方法都能访问
        self.exit = False
        
        # 创建bash进程，设置text=True自动处理字符串编码
        self.process1 = subprocess.Popen(
            ["bash"],
            stderr=subprocess.PIPE,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            text=True
        )
        self.process2 = subprocess.Popen(
            ["bash"],
            stderr=subprocess.PIPE,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            text=True
        )
        
        # 启动监控线程，设置daemon=True随主程序退出
        threading.Thread(target=self.read_stdout, args=(self.process1, "process_1"), daemon=True).start()
        threading.Thread(target=self.read_stderr, args=(self.process1, "process_1"), daemon=True).start()
        threading.Thread(target=self.read_stdout, args=(self.process2, "process_2"), daemon=True).start()
        threading.Thread(target=self.read_stderr, args=(self.process2, "process_2"), daemon=True).start()
        
        self.frame2 = tk.Frame(self.ws, bg="#777474")
        self.frame2.grid(column=0, row=0, padx=(100,100), pady=(10, 10), sticky="W")
        
        self.btnVerify = tk.Button(self.ws, text='Start', command=self.start_execution)
        self.btnVerify.grid(row=5, columnspan=3, pady=10)
        
        self.logArea = tkst.ScrolledText(self.ws, wrap=tk.WORD, width=80, height=20)
        self.logArea.grid(padx=(100,100), pady=(10, 10), row=10, sticky="W")

    def safe_insert(self, text):
        """线程安全的文本插入方法，通过after在主线程执行"""
        self.ws.after(0, lambda: self.logArea.insert(tk.INSERT, text + "\n"))
        self.ws.after(0, lambda: self.logArea.see(tk.END))

    def read_stdout(self, process, process_name):
        while not self.exit:
            msg = process.stdout.readline()
            if msg:
                self.safe_insert(f"{process_name} stdout: {msg.strip()}")

    def read_stderr(self, process, process_name):
        while not self.exit:
            msg = process.stderr.readline()
            if msg:
                self.safe_insert(f"{process_name} ERROR: {msg.strip()}")

    def execute_next_step(self):
        """分步执行命令，避免阻塞主线程"""
        if self.flag == 1:
            print("Case 1: Executing test1.py")
            self.process1.stdin.write(r'python C:\path\test1.py' + "\n")
            self.process1.stdin.flush()
            self.process1.wait()  # 等待当前命令执行完成，可根据需求移除
            self.flag = 2
            self.ws.after(100, self.execute_next_step)
        elif self.flag == 2:
            print("Case 2: Executing test2.py")
            self.process1.stdin.write(r'python C:\path\test2.py' + "\n")
            self.process1.stdin.flush()
            self.process1.wait()
            self.flag = 3
            self.ws.after(100, self.execute_next_step)
        elif self.flag == 3:
            print("Case 3: Executing test3.py")
            self.process2.stdin.write(r'python C:\path\test3.py' + "\n")
            self.process2.stdin.flush()
            self.process2.wait()
            self.flag = 4
            self.ws.after(100, self.execute_next_step)
        elif self.flag == 4:
            print("Case 4: Executing test4.py")
            self.process2.stdin.write(r'python C:\path\test4.py' + "\n")
            self.process2.stdin.flush()
            self.process2.wait()
            self.flag = 5
            self.safe_insert("所有命令执行完成")
        else:
            return

    def start_execution(self):
        print("Button clicked")
        self.flag = 1
        # 启动分步执行，避免阻塞主线程
        self.ws.after(0, self.execute_next_step)

    def run(self):
        self.ws.title('Test Script')
        # 窗口关闭时设置exit标志，终止监控线程
        self.ws.protocol("WM_DELETE_WINDOW", lambda: setattr(self, "exit", True) or self.ws.destroy())
        self.ws.mainloop() 
        
validator = TKValidator()
validator.run() 

关键修改说明 #

• 变量作用域修正：将exit改为实例变量self.exit，确保所有方法都能访问；用普通整数变量替代IntVar，简化逻辑。
• bash命令修复：写入命令时添加换行符\n，确保bash执行命令；设置text=True让subprocess直接处理字符串，无需手动编码解码。
• 线程安全GUI操作：新增safe_insert方法，通过self.ws.after()在主线程中执行文本插入，避免线程冲突。
• 流程控制优化：将原ButtonClick中的while循环改为execute_next_step分步执行，用after()触发下一步，避免阻塞主线程；添加process.wait()等待每个命令执行完成（可根据需求移除，实现异步执行）。
• 线程守护与退出处理：设置线程为daemon=True，确保主程序退出时线程自动终止；窗口关闭时设置self.exit=True，终止监控线程。

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