Matplotlib多线程绘图与输入读取异常问题求助

核心问题：Matplotlib GUI后端的线程安全限制 #

Matplotlib的所有GUI后端（如TkAgg、Qt5Agg等）不支持跨线程操作，所有涉及绘图窗口渲染、更新的操作必须在创建绘图窗口的主线程中执行。同时，GUI需要持续运行自己的事件循环来处理界面刷新、用户交互，一旦主线程被阻塞（比如input()、time.sleep()），事件循环无法运行，图表就会停止更新甚至无法显示。

第一个案例失败原因 #

# 原代码核心逻辑
with concurrent.futures.ThreadPoolExecutor() as executor:
    plot_thread = executor.submit(plotter, ax, voltage)  # 子线程直接操作GUI元素

for _ in range(15):
    res = input("Type: ")  # 主线程被阻塞，GUI事件循环停摆

• 子线程直接修改ax、canvas等GUI组件，违反Matplotlib线程安全规则，绘图更新无法被GUI后端正确处理。
• 主线程被input()阻塞，导致Matplotlib的GUI事件循环无法运行，即使子线程调用了draw()和flush_events()，界面也无法触发刷新。

第二个案例失败原因 #

# 原代码核心逻辑
with concurrent.futures.ThreadPoolExecutor() as executor:
    new_thread = executor.submit(fetch_input)  # 子线程处理终端输入

while True:
    # 主线程绘图，但事件循环被终端输入干扰
    plt.pause(1)

• input()是阻塞式操作，终端的输入输出通常由主线程控制，子线程调用input()会抢占终端焦点，导致主线程中plt.pause(1)触发的临时事件循环无法正常启动，直到子线程的输入循环结束，图表才会显示。
• 虽然绘图逻辑放在了主线程，但子线程的input()阻塞了终端交互流程，延迟了fig.show()的界面初始化。

可行解决方案 #

方案1：用Matplotlib动画模块管理动态绘图

利用FuncAnimation让Matplotlib自主管理事件循环，避免手动线程操作：

import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation
import threading

voltage = []
stop_flag = False

fig, ax = plt.subplots()
line, = ax.plot(voltage)

def update(frame):
    if stop_flag:
        ani.event_source.stop()
        print("stream interrupted")
        return line,
    voltage.append(1)
    line.set_data(range(len(voltage)), voltage)
    ax.relim()
    ax.autoscale_view()
    return line,

def check_input():
    global stop_flag
    while not stop_flag:
        res = input("Type: ")
        if res == "q":
            stop_flag = True
            break
        print(res)

# 子线程仅处理输入，不触碰GUI
input_thread = threading.Thread(target=check_input)
input_thread.start()

# 主线程运行动画（自动维护GUI事件循环）
ani = FuncAnimation(fig, update, interval=1000, blit=True)
plt.show()

方案2：主线程运行GUI事件循环，子线程处理输入

确保所有GUI操作在主线程执行，子线程仅修改状态变量，主线程定期检查状态并更新图表：

import matplotlib.pyplot as plt
import threading
import time

voltage = []
stop_flag = False

def fetch_input():
    global stop_flag
    for _ in range(15):
        res = input("Type: ")
        if res == "q":
            stop_flag = True
            break
        print(res)

# 启动输入处理子线程（守护线程随主线程退出）
threading.Thread(target=fetch_input, daemon=True).start()

# 主线程初始化绘图并维持事件循环
fig, ax = plt.subplots()
plt.show(block=False)

while not stop_flag:
    voltage.append(1)
    ax.clear()
    ax.plot(voltage)
    fig.canvas.draw()
    fig.canvas.flush_events()
    time.sleep(1)

print("main loop interrupted")

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