1. 为什么实时绘图不是“定时刷新”那么简单很多人第一次接触 PyQt5 加 pyqtgraph 做实时数据绘制脑子里冒出来的方案是开一个QTimer每 100 毫秒把新数据塞进列表然后调用一次plot()重画。这个思路在数据量小、刷新频率低的时候确实能跑起来但只要采样率上到几百赫兹、或者窗口里同时挂了三五条曲线界面就会开始卡顿、鼠标拖动窗口时曲线直接冻结严重的时候整个程序无响应。问题的根子在于实时绘图的核心矛盾不是“怎么画”而是“怎么在不阻塞主线程的前提下把数据高效地送进绘图区”。PyQt5 的所有界面刷新都发生在主线程也叫 GUI 线程而数据采集往往来自串口、网口、传感器回调或者一个独立的工作线程。如果你在采集线程里直接操作绘图对象轻则画面撕裂重则直接崩溃——Qt 的绘图对象不是线程安全的。pyqtgraph 之所以在实时绘图场景里比 matplotlib 更合适是因为它底层用了 Qt 的QGraphicsScene体系并且针对“频繁更新少量数据”做了专门优化。它提供了一个关键类PlotDataItem配合setData()方法可以在不重建整个图元的情况下只更新数据点。这一点和 matplotlib 每次draw()都要重绘整个画布有本质区别。所以这篇内容我想聊的不是“怎么装库、怎么画一条线”这种入门操作而是围绕一个真实的实时监测场景把线程模型、数据缓冲、刷新策略、性能调优这几件事串起来讲清楚。适合已经能跑通 pyqtgraph 基础示例、但在实际项目里遇到卡顿或数据错乱的开发者。如果你正在做工业监测、传感器波形显示、音频可视化或者任何需要“数据来了就画”的桌面工具下面的内容应该能帮你少走不少弯路。2. 线程模型选型QThread、moveToThread 还是信号槽2.1 三种常见方案的取舍逻辑在 PyQt5 里做后台采集绕不开线程。我见过最多的三种写法是继承QThread重写run()、用moveToThread()把工作对象挪到子线程、以及直接用 Python 的threading.Thread。这三种都能跑但在实时绘图场景下表现差别很大。继承QThread重写run()是最直观的但问题在于QThread对象本身属于创建它的线程通常是主线程只有run()里的代码跑在新线程。很多人会把槽函数也写在QThread子类里然后惊讶地发现槽函数还是在主线程执行——因为对象归属没变。这个坑非常经典。moveToThread()是 Qt 官方更推荐的方式定义一个普通的QObject工作类在里面写采集逻辑和槽函数然后创建一个QThread实例把工作对象moveToThread(thread)。这样工作对象的所有槽函数都会在子线程的事件循环里执行信号槽跨线程时 Qt 会自动用队列连接QueuedConnection数据传递天然安全。直接用threading.Thread也能用但你就得自己处理线程间通信而且没法享受 Qt 信号槽的自动排队机制。在纯 Python 层面传数据给 GUI 时必须借助信号或者QMetaObject.invokeMethod否则就是在子线程碰 GUI 对象风险很高。我的建议很明确用moveToThread()方案。它既符合 Qt 的线程规范又能让信号槽自动跨线程排队代码结构也清晰。2.2 一个可复用的采集线程骨架下面这个骨架是我在多个项目里反复用过的采集逻辑放在Worker里通过信号把数据块发回主线程from PyQt5.QtCore import QObject, QThread, pyqtSignal, QTimer import numpy as np class DataWorker(QObject): # 发送一个数据块用 numpy 数组承载避免 Python list 开销 chunk_ready pyqtSignal(object) finished pyqtSignal() def __init__(self, interval_ms10, chunk_size50): super().__init__() self.interval_ms interval_ms self.chunk_size chunk_size self._running False def start(self): self._running True self._timer QTimer() self._timer.setInterval(self.interval_ms) self._timer.timeout.connect(self._produce) self._timer.start() def _produce(self): if not self._running: return # 这里替换成真实采集逻辑比如读串口、读寄存器 t np.linspace(0, 1, self.chunk_size) data np.sin(2 * np.pi * 5 * t) np.random.normal(0, 0.05, self.chunk_size) self.chunk_ready.emit(data) def stop(self): self._running False if hasattr(self, _timer): self._timer.stop() self.finished.emit()注意几个细节chunk_ready信号携带的是object类型因为 pyqtgraph 和 numpy 配合时用数组比用 list 快得多QTimer是在子线程里创建的它的timeout会在子线程事件循环里触发不会干扰主线程_running标志位用来安全停止避免定时器还在跑但对象已经被销毁。主线程这边这样接self.thread QThread() self.worker DataWorker(interval_ms10, chunk_size50) self.worker.moveToThread(self.thread) self.thread.started.connect(self.worker.start) self.worker.chunk_ready.connect(self.on_chunk) self.thread.start()on_chunk是主线程里的槽函数收到数据后负责更新绘图。因为信号槽是队列连接数据会先进入主线程的事件队列等主线程空闲时再处理天然避免了竞态。2.3 为什么不用 QThread 的 run 里直接 while 循环有人喜欢在run()里写while True: data read(); emit(data)。这种写法在采集频率很高时确实能减少定时器开销但它有个致命问题没有事件循环槽函数无法在该线程执行。如果你后续想给这个线程发停止信号信号会排队但永远不被处理只能靠标志位轮询。而且while循环里如果emit太快主线程事件队列会堆积界面反而更卡。用QTimer驱动的好处是节奏可控每次timeout之间线程有机会处理其他事件比如停止信号而且可以通过调整interval_ms和chunk_size来平衡延迟与吞吐。实测下来10 毫秒发 50 个点相当于 5000 点/秒的吞吐对 pyqtgraph 来说完全在舒适区。3. 数据缓冲策略环形缓冲与增量更新3.1 为什么不能无限 append最朴素的写法是维护一个 Python list每次收到新数据就extend然后setData(x, y)全量重画。数据量小的时候没问题但当 list 涨到几十万个点每次setData都要把整个数组传给绘图项CPU 时间会线性增长内存也会持续膨胀。实时监测场景通常只关心最近一段时间窗口比如最近 10 秒的波形老数据没必要一直留着。这时候就需要环形缓冲ring buffer。它的思路是预先分配一个固定长度的 numpy 数组用写指针循环覆盖。这样内存恒定而且切片操作是 O(1) 的视图操作不涉及数据搬移。class RingBuffer: def __init__(self, capacity): self.capacity capacity self.buffer np.zeros(capacity, dtypenp.float32) self.write_pos 0 self.size 0 def append(self, data): n len(data) if n self.capacity: # 数据比缓冲区还大只保留最后 capacity 个 self.buffer[:] data[-self.capacity:] self.write_pos 0 self.size self.capacity return end self.write_pos n if end self.capacity: self.buffer[self.write_pos:end] data else: # 绕回开头 first self.capacity - self.write_pos self.buffer[self.write_pos:] data[:first] self.buffer[:n - first] data[first:] self.write_pos end % self.capacity self.size min(self.size n, self.capacity) def get_ordered(self): if self.size self.capacity: return self.buffer[:self.size] # 已满从 write_pos 开始读才是时间顺序 return np.concatenate((self.buffer[self.write_pos:], self.buffer[:self.write_pos]))get_ordered里那个np.concatenate在缓冲区满的时候每次都要拷贝一次如果每帧都调用会有开销。优化办法是维护两个视图分别传给 pyqtgraph或者干脆接受这一次拷贝——实测 10 万个 float32 的拼接大约 0.1 毫秒对 60 帧刷新来说可以忽略。3.2 增量更新与全量更新的分界线pyqtgraph 的setData支持两种模式全量替换和增量追加。很多人不知道setData还有setData(x, y)之外的用法其实它内部会做差分判断但前提是你传的数组对象有稳定的引用关系。更高效的做法是当新数据只是追加到末尾时用setData传完整数组当数据窗口滑动时用切片视图。pyqtgraph 对 numpy 视图的支持很好只要 dtype 和形状对得上它不会强制拷贝。但这里有个陷阱如果你传的是环形缓冲的get_ordered()结果每次都是新数组pyqtgraph 无法复用之前的顶点缓冲会重新构建。解决办法是维护一个固定长度的显示数组每次把新数据写进去然后setData传这个数组的引用。虽然还是全量但省去了拼接开销。我一般会设一个阈值窗口内点数少于 5000 时全量更新超过 5000 时考虑降采样或者只更新可见区域。pyqtgraph 本身有setDownsampling和setClipToView两个方法开启后它会自动根据视图范围做抽稀对大数据量帮助很大。curve plot.plot(peny) curve.setDownsampling(autoTrue, methodpeak) curve.setClipToView(True)methodpeak会保留每段区间的最大值和最小值波形不会因为抽稀而丢失尖峰这对监测类应用很重要。3.3 时间轴的处理技巧实时绘图通常横轴是时间。如果直接用time.time()作为 x 值数值会很大十亿级别pyqtgraph 的坐标轴标签会显示一长串数字可读性差。更好的做法是用相对时间记录一个起始时间戳x 轴传current - start单位秒。但这样又带来一个问题当程序运行几小时后x 值会变得很大浮点精度下降曲线可能出现抖动。解决办法是定期重置基准时间或者用int毫秒计数配合自定义AxisItem格式化显示。我通常会在on_chunk里这样处理def on_chunk(self, data): n len(data) now time.perf_counter() t self.t0 np.arange(self.sample_count, self.sample_count n) / self.fs self.sample_count n self.ring_x.append(t) self.ring_y.append(data) self.curve.setData(self.ring_x.get_ordered(), self.ring_y.get_ordered())用采样计数除以采样率来生成时间轴比用time.time()更稳定不受系统时钟调整影响。perf_counter只用来做性能测量不参与坐标计算。4. 刷新节奏控制定时器、帧率与背压4.1 采集频率和刷新频率必须解耦新手最容易犯的错误是采集多快就画多快。传感器 1kHz 采样就每毫秒调一次setData。结果就是主线程被绘图占满界面完全没法交互。正确的做法是采集和刷新解耦采集线程按自己的节奏往缓冲区写数据主线程用一个独立的QTimer按固定帧率比如 30fps 或 60fps从缓冲区取最新数据刷新画面。这样无论采集多快界面刷新始终稳定。self.render_timer QTimer() self.render_timer.setInterval(33) # 约 30fps self.render_timer.timeout.connect(self.render) self.render_timer.start()render函数里只做一件事从环形缓冲取当前窗口的数据调setData。因为缓冲区是线程安全的只有采集线程写、主线程读且写的是 numpy 数组的不同位置不需要加锁。4.2 背压当渲染跟不上采集时怎么办如果采集速率远高于渲染速率缓冲区会被快速覆盖但这不是问题——因为实时监测本来就只关心最新数据老数据被覆盖是符合预期的。真正的问题是事件队列积压如果采集线程emit信号的速度超过主线程处理速度chunk_ready信号会在主线程事件队列里堆积导致界面延迟越来越大。解决办法有两个一是降低emit频率把多个小数据块合并成一个大块再发二是在主线程用QApplication.processEvents()主动处理队列但这会带来重入风险不推荐。我更倾向于第一种采集线程内部维护一个小缓冲每积累 N 个点或者每过 M 毫秒才emit一次。这样信号数量可控主线程压力小。比如 1kHz 采集每 20 毫秒发一次每次 20 个点信号频率降到 50Hz完全在可控范围。4.3 用 QElapsedTimer 做自适应帧率固定帧率在大多数场景够用但如果数据量突然增大比如用户把时间窗口从 5 秒拉到 60 秒固定 30fps 可能导致每帧渲染时间超过 33 毫秒界面开始卡。这时候可以做一个简单的自适应测量上一帧的渲染耗时如果超过阈值就降低帧率。self.elapsed QElapsedTimer() def render(self): self.elapsed.start() # ... 更新绘图 ... cost self.elapsed.elapsed() if cost 40: self.render_timer.setInterval(min(100, self.render_timer.interval() 5)) elif cost 15 and self.render_timer.interval() 16: self.render_timer.setInterval(max(16, self.render_timer.interval() - 2))这段逻辑的意思是渲染超过 40 毫秒就降帧低于 15 毫秒就尝试升帧但最低不低于 16 毫秒约 60fps。实测下来这种自适应策略在窗口大小变化、数据量波动时表现很稳用户几乎感觉不到帧率变化。注意QElapsedTimer的elapsed()返回毫秒整数精度足够。不要用time.time()做帧耗时测量系统时钟调整会影响结果。5. 绘图性能调优从 5000 点到 50 万点5.1 关闭不必要的视觉特性pyqtgraph 默认开启了一些方便但耗性能的特性实时绘图时应该按需关闭特性默认状态实时绘图建议原因抗锯齿开启关闭每条线每个点都要做混合计算开销大鼠标交互开启保留但限制缩放平移有用但autoRange频繁触发会重算网格开启按需网格线多了会增加图元数量图例关闭按需图例更新会触发额外布局数据点标记关闭关闭每个点画一个符号点数多时灾难关闭抗锯齿的代码pg.setConfigOptions(antialiasFalse)这一行放在创建PlotWidget之前。实测在 10 万点场景下关闭抗锯齿能提升约 30% 的渲染速度。5.2 用 OpenGL 加速如果数据量真的很大几十万点以上可以考虑开启 pyqtgraph 的 OpenGL 模式pg.setConfigOptions(useOpenGLTrue)但 OpenGL 模式有兼容性问题某些集成显卡或者远程桌面环境下会出问题而且开启后setData的某些优化会失效。我的建议是先做 CPU 侧优化实在不够再上 OpenGL。大多数工业监测场景10 万点以内用 CPU 渲染完全够用。5.3 多曲线场景的图元管理一个窗口里画多条曲线时每条曲线都是一个独立的PlotDataItem。如果曲线数量超过 10 条图元管理本身就会成为瓶颈。这时候可以考虑把不常看的曲线隐藏setVisible(False)而不是删除重建用PlotCurveItem替代PlotDataItem前者更轻量但不支持符号标记多条曲线共用同一个 x 轴数组减少内存和传输开销# 多条曲线共享 x 轴 x self.ring_x.get_ordered() for i, curve in enumerate(self.curves): curve.setData(x, self.ring_y[i].get_ordered())共享 x 轴数组的好处是 pyqtgraph 内部可以复用顶点缓冲的 x 分量只更新 y 分量。虽然官方文档没明说但实测在 5 条曲线场景下能省 15% 左右的渲染时间。5.4 避免在绘图循环里做重计算我见过有人在render函数里做 FFT、滤波、归一化然后才画图。这些计算如果每帧都做会严重拖慢刷新。正确做法是计算放在采集线程或者独立的计算线程主线程只负责把算好的数据画出来。如果确实需要在主线程做轻量处理尽量用 numpy 的向量化操作避免 Python 循环。比如归一化# 慢 y [(v - min_val) / (max_val - min_val) for v in data] # 快 y (data - data.min()) / (data.max() - data.min() 1e-9)numpy 的向量化在 10 万点级别能比 Python 循环快几十倍这个差距在实时场景里就是流畅和卡顿的区别。6. 踩过的坑与排查链路6.1 界面冻结从现象到根因的完整排查现象程序运行几分钟后界面完全无响应鼠标点不动但数据还在采集日志还在输出。排查第一步确认是主线程被阻塞还是死锁。用py-spy dump --pid pid抓一下主线程栈如果看到栈顶停在setData或者paintEvent说明是渲染太慢如果停在threading.Lock.acquire说明是死锁。排查第二步如果是渲染慢检查是不是在render里做了全量setData且数据量巨大。我遇到过一次是环形缓冲的get_ordered在缓冲区满时每次concatenate10 万点30fps 下每秒拼接 300 万点CPU 直接跑满。修复改成维护两个视图或者用np.roll预分配。最终方案是显示数组固定长度采集线程直接往显示数组的对应位置写主线程只传引用省去拼接。排查第三步如果栈显示在QApplication.exec_但界面不刷新可能是事件循环被某个模态对话框或者processEvents递归阻塞了。检查有没有在槽函数里弹QMessageBox或者调用exec_。6.2 数据错乱跨线程访问的隐蔽表现现象曲线偶尔出现一段乱跳的数据或者 x 轴和 y 轴长度不一致导致崩溃。根因采集线程在写 numpy 数组的同时主线程在读同一个数组。numpy 的切片赋值不是原子操作如果主线程在采集线程写一半时读取就会拿到不一致的数据。修复用双缓冲。采集线程写 buffer A写完后通过信号把 buffer A 的引用发给主线程同时切换到 buffer B 继续写。主线程收到信号后读 buffer A读完通知采集线程可以复用。这样读写分离没有竞态。class DoubleBuffer: def __init__(self, capacity): self.buffers [RingBuffer(capacity), RingBuffer(capacity)] self.active 0 def write(self, data): self.buffers[self.active].append(data) def swap(self): self.active 1 - self.active return self.buffers[1 - self.active]采集线程每积累一批数据就swap一次并emit主线程拿到的是已经写完的那个缓冲区不会读到半截数据。6.3 内存泄漏曲线越跑越卡现象程序启动时流畅运行几小时后越来越卡内存占用持续上升。排查用tracemalloc或者objgraph看对象数量。常见原因是每次render都创建新的PlotDataItem或者新的 numpy 数组旧对象没有被回收。修复绘图项只创建一次后续只调setData。numpy 数组用预分配的不要每帧np.array(list)。另外注意信号连接如果每次采集都connect一次槽函数会导致一个信号触发多次槽而且旧连接不会自动断开。# 错误每次采集都连接 def start_acquisition(self): self.worker.chunk_ready.connect(self.on_chunk) # 重复连接 # 正确只连接一次 def __init__(self): self.worker.chunk_ready.connect(self.on_chunk)6.4 停止时的崩溃线程未正确退出现象关闭窗口时程序崩溃报QThread: Destroyed while thread is still running。根因窗口关闭时QThread还在运行Python 垃圾回收把QThread对象销毁了但底层线程还在跑。修复在closeEvent里先停止采集等线程真正退出再关闭。def closeEvent(self, event): self.worker.stop() self.thread.quit() self.thread.wait(2000) # 最多等 2 秒 event.accept()wait的返回值要检查如果返回False说明线程没在超时内退出可能需要强制terminate()但那是最后手段正常情况应该通过标志位让线程自己结束。7. 一个完整的可运行示例结构把上面的要点串起来一个典型的实时绘图程序结构是这样的import sys import time import numpy as np import pyqtgraph as pg from PyQt5.QtCore import QObject, QThread, QTimer, pyqtSignal from PyQt5.QtWidgets import QApplication, QMainWindow, QVBoxLayout, QWidget pg.setConfigOptions(antialiasFalse) class RingBuffer: # 省略见第 3 节 pass class DataWorker(QObject): chunk_ready pyqtSignal(object) def __init__(self, fs1000, chunk_ms20): super().__init__() self.fs fs self.chunk_size int(fs * chunk_ms / 1000) self._running False def start(self): self._running True self._timer QTimer() self._timer.setInterval(20) self._timer.timeout.connect(self._produce) self._timer.start() def _produce(self): if not self._running: return t np.arange(self.chunk_size) / self.fs data np.sin(2 * np.pi * 10 * t) np.random.normal(0, 0.1, self.chunk_size) self.chunk_ready.emit(data.astype(np.float32)) def stop(self): self._running False if hasattr(self, _timer): self._timer.stop() class MainWindow(QMainWindow): def __init__(self): super().__init__() self.fs 1000 self.window_sec 5 self.capacity self.fs * self.window_sec self.ring_y RingBuffer(self.capacity) self.ring_x RingBuffer(self.capacity) self.sample_count 0 self.plot_widget pg.PlotWidget() self.plot_widget.setYRange(-2, 2) self.curve self.plot_widget.plot(penpg.mkPen(y, width1)) self.curve.setDownsampling(autoTrue, methodpeak) self.curve.setClipToView(True) self.setCentralWidget(self.plot_widget) self.resize(1000, 500) self.thread QThread() self.worker DataWorker(fsself.fs) self.worker.moveToThread(self.thread) self.thread.started.connect(self.worker.start) self.worker.chunk_ready.connect(self.on_chunk) self.thread.start() self.render_timer QTimer() self.render_timer.setInterval(33) self.render_timer.timeout.connect(self.render) self.render_timer.start() def on_chunk(self, data): n len(data) t np.arange(self.sample_count, self.sample_count n) / self.fs self.sample_count n self.ring_x.append(t.astype(np.float32)) self.ring_y.append(data) def render(self): x self.ring_x.get_ordered() y self.ring_y.get_ordered() if len(x) len(y) and len(x) 0: self.curve.setData(x, y) def closeEvent(self, event): self.worker.stop() self.thread.quit() self.thread.wait(2000) event.accept() if __name__ __main__: app QApplication(sys.argv) win MainWindow() win.show() sys.exit(app.exec_())这个结构跑 1kHz 采样、5 秒窗口、30fps 刷新CPU 占用在普通笔记本上大约 5% 到 8%界面交互完全流畅。你可以把DataWorker._produce里的模拟数据换成真实的串口读取、网口接收或者传感器回调其余部分不用动。8. 几个容易被忽略的细节第一setData的clear参数。pyqtgraph 的setData默认会清空旧数据再设置新的但如果你传的数组长度和之前一样它内部会走快速路径。如果长度变化频繁可以显式传clearFalse试试但要注意旧数据可能残留。这个参数的行为在不同版本里有差异建议以实测为准。第二坐标轴自动范围的开销。PlotWidget默认开启autoRange每次setData后都会重新计算数据范围。如果数据范围稳定手动setYRange和setXRange能省不少时间。我通常会在初始化时设好范围只在用户主动缩放时才让 pyqtgraph 接管。第三高 DPI 屏幕的适配。在 4K 屏上pyqtgraph 默认的线宽和字体可能偏小。可以在QApplication创建前设置QApplication.setAttribute(Qt.AA_EnableHighDpiScaling)然后调整pg.setConfigOptions(foregroundw, backgroundk)来适配深色主题。第四打包发布时的坑。用 PyInstaller 打包 pyqtgraph 程序时需要把 pyqtgraph 的资源文件一起打进去否则运行时可能报找不到图标或者样式。在 spec 文件里加datas[(pg.__path__[0], pyqtgraph)]能解决大部分问题。第五多显示器场景。如果程序窗口被拖到不同刷新率的显示器之间QTimer的帧率可能和显示器不同步导致画面撕裂。这种情况可以考虑用QScreen.refreshRate()动态调整渲染定时器间隔但大多数监测场景对撕裂不敏感可以忽略。整体来说PyQt5 加 pyqtgraph 做实时绘图难点不在 API 调用而在线程边界划分、数据缓冲设计和刷新节奏控制这三件事上。把这三件事理顺了剩下的就是按需调参。我在实际项目里最深的体会是不要等到卡顿了才去优化一开始就把环形缓冲和采集渲染分离的架子搭好后面无论数据量怎么涨改起来都只是调几个参数的事。
