如何在Flask多线程环境中使用Pyqtgraph绘图并解决线程报错？

你遇到的核心问题是PyQt5的QApplication及相关组件（包括PyQtGraph依赖的定时器、绘图对象）必须绑定到由QThread创建的单一线程，而Flask默认的多线程请求模型中，每个请求会启动普通Python线程（并非QThread），第二次请求时在新线程中复用或创建PyQt对象就会触发定时器相关报错。

核心思路 #

将所有PyQtGraph绘图操作隔离到一个单独的QThread线程中，让QApplication始终运行在这个专属线程里，Flask的请求线程通过队列传递绘图任务，避免在非QThread线程中操作PyQt对象。

具体实现步骤 #

1. 初始化全局QApplication与任务队列：在Flask启动时，创建后台QThread线程专门运行QApplication和处理绘图任务，用队列实现请求线程与QThread线程的通信。
2. 封装绘图任务：把绘图需要的参数（如数据、标题、输出路径）打包成任务对象，放入队列由QThread线程取出执行。
3. 线程安全的结果返回：绘图完成后，将生成的图片信息放回结果队列，请求线程等待结果后返回给前端。

代码示例 #

import sys
from queue import Queue
from flask import Flask, send_file
import pyqtgraph as pg
import pandas as pd
from PyQt5.QtWidgets import QApplication
from PyQt5.QtCore import QThread, QObject, pyqtSignal, pyqtSlot

app = Flask(__name__)

# 全局任务队列：传递绘图请求
task_queue = Queue()
# 全局结果队列：返回绘图结果
result_queue = Queue()

class PlotWorker(QObject):
    """处理绘图任务的工作对象，运行在QThread中"""
    finished = pyqtSignal()

    @pyqtSlot()
    def run(self):
        # 初始化QApplication，仅在当前QThread中执行一次
        if not QApplication.instance():
            qapp = QApplication(sys.argv)
        
        while True:
            # 从队列获取任务
            task = task_queue.get()
            if task is None:  # 终止信号
                break
            
            title, df, tmpfile = task
            try:
                # 所有PyQtGraph操作都在这个QThread中执行
                pg_layout = pg.GraphicsLayoutWidget(title=title, size=(2880, 960))
                p1 = pg_layout.addPlot(row=0, col=0)
                p1.plot(df["C"], pen=pg.mkPen('g'))
                
                # 导出图片
                exporter = pg.exporters.ImageExporter(pg_layout.sceneObj)
                exporter.export(tmpfile)
                
                # 返回成功结果
                result_queue.put((True, tmpfile))
            except Exception as e:
                # 返回错误信息
                result_queue.put((False, str(e)))
            finally:
                task_queue.task_done()
        
        self.finished.emit()

# 启动绘图线程
def start_plot_thread():
    thread = QThread()
    worker = PlotWorker()
    worker.moveToThread(thread)
    thread.started.connect(worker.run)
    worker.finished.connect(thread.quit)
    worker.finished.connect(worker.deleteLater)
    thread.finished.connect(thread.deleteLater)
    thread.start()
    return thread

# 启动Flask前初始化绘图线程
plot_thread = start_plot_thread()

@app.route('/plot')
def plot():
    # 模拟数据，实际应从请求参数或数据库获取
    df = pd.DataFrame({"C": [1,3,2,5,4]})
    tmpfile = "/tmp/plot.png"
    title = "Test Plot"
    
    # 发送绘图任务到队列
    task_queue.put((title, df, tmpfile))
    # 等待结果
    success, result = result_queue.get()
    
    if success:
        return send_file(tmpfile, mimetype='image/png')
    else:
        return f"绘图失败: {result}", 500

if __name__ == '__main__':
    try:
        app.run(threaded=True)
    finally:
        # 停止绘图线程
        task_queue.put(None)
        plot_thread.wait()

关键注意事项 #

• QApplication仅初始化一次：必须在QThread线程内创建QApplication，且确保全局只有一个实例。
• 所有PyQt操作必须在QThread中执行：包括创建GraphicsLayoutWidget、绘图、导出图片，不能在Flask的请求线程中操作这些对象。
• 使用队列实现线程通信：避免直接在多线程间共享PyQt对象，通过队列传递任务和结果，保证线程安全。

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