Python用Paramiko SSH执行MQTT命令取输出及Tkinter卡顿问题求解

问题根因 #

• 执行逻辑顺序错误：当前代码先同步调用subprocess.call运行TEST发布脚本，等待脚本完全执行结束后才启动CHECK订阅逻辑，TEST发布的MQTT消息早已丢失，自然订阅端无任何输出，导致stdout.read()永久阻塞。
• Tkinter界面阻塞：Tkinter是单线程事件循环模型，所有阻塞式操作（SSH连接、订阅等待、IO读取）直接在主线程运行时会卡住界面刷新，出现无响应状态。

解决方案 #

核心调整点 #

1. 调整执行顺序：先启动MQTT订阅，等待1~2秒确保订阅生效后，再执行发布命令，保证消息能被捕获。
2. 子线程隔离阻塞逻辑：将整个测试流程放到独立子线程运行，不占用Tkinter主线程的事件循环。
3. 实时读取订阅输出：逐行读取订阅命令的标准输出，匹配到测试通过的结果后主动终止订阅进程，无需手动按CTRL+C。
4. 无需拆分两个脚本，同一进程内建立两个SSH连接即可，减少不必要的进程调度开销。

修改后可运行代码 #

import threading
import time
import tkinter as tk
import paramiko

hostname = "xxx.xxx.xxx.xxx"
username = "xxx"
password = "xxx"

def test_flow():
    # 第一步：先建立订阅用的SSH连接，启动CHECK命令
    client_sub = paramiko.SSHClient()
    client_sub.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        client_sub.connect(hostname=hostname, username=username, password=password)
        print('订阅端SSH连接成功')
    except Exception as e:
        print(f"[!] 订阅端无法连接SSH服务器: {str(e)}")
        return

    # 启动订阅命令，get_pty=True保证可以发送中断信号
    stdin_sub, stdout_sub, stderr_sub = client_sub.exec_command(
        'mosquitto_sub -t "/test/#" -v',
        get_pty=True
    )

    # 等待订阅生效
    time.sleep(1.5)

    # 第二步：建立发布用的SSH连接，执行TEST命令
    client_pub = paramiko.SSHClient()
    client_pub.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        client_pub.connect(hostname=hostname, username=username, password=password)
        print('发布端SSH连接成功')
    except Exception as e:
        print(f"[!] 发布端无法连接SSH服务器: {str(e)}")
        # 失败了先关闭订阅连接再返回
        client_sub.close()
        return

    # 执行发布命令
    pub_cmd = "mosquitto_pub -t \"/test/start\" -m '{\"type\": \"phcomtest\", \"cycles\": 3}'"
    stdin_pub, stdout_pub, stderr_pub = client_pub.exec_command(pub_cmd)
    err_pub = stderr_pub.read().decode()
    if err_pub:
        print(f"发布命令执行错误: {err_pub}")
        client_pub.close()
        client_sub.close()
        return
    print("TEST消息发布完成，等待订阅结果...")
    client_pub.close()

    # 第三步：逐行读取订阅输出，匹配测试结果
    test_pass = False
    # 这里可以根据你的实际通过输出修改匹配规则
    pass_keyword = "test pass" # 替换为你实际的通过标识字符串
    try:
        # 设置超时时间，避免永久等待，比如测试最长跑30秒就超时
        timeout = 30
        start_time = time.time()
        while time.time() - start_time < timeout:
            if stdout_sub.channel.exit_status_ready():
                break
            # 读一行输出
            line = stdout_sub.readline()
            if not line:
                time.sleep(0.1)
                continue
            print(f"收到订阅输出: {line.strip()}")
            if pass_keyword in line:
                test_pass = True
                break
    finally:
        # 主动发送CTRL+C终止订阅命令
        stdin_sub.write('\x03')
        stdin_sub.flush()
        client_sub.close()

    if test_pass:
        print("✅ 测试通过")
    else:
        print("❌ 测试未通过或超时")

def initiate():
    # 把测试流程放到子线程运行，不卡界面
    threading.Thread(target=test_flow, daemon=True).start()

# 界面代码不变
window = tk.Tk()
window.title("MQTT")
window.geometry("300x300")
window.resizable(False, False)

window.rowconfigure([0], minsize=300, weight=0)
window.columnconfigure([0], minsize=300, weight=0)

btn_abl = tk.Button(window, text="Start", command = initiate)
btn_abl.grid(row=0, column=0, sticky="nsew")

window.mainloop()

补充说明 #

• 代码中的pass_keyword需要替换为你实际测试通过时订阅端输出的特征字符串，匹配到就会判定通过。
• 超时时间默认30秒，可根据TEST实际运行时长调整。
• 所有SSH连接用完都会自动关闭，不会残留连接。

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