上位机数据采集、储存与实时显示：从串口到SQLite的完整实现

简介：一套面向工业自动化、物联网与嵌入式场景的C#上位机数据采集、存储与实时显示工程，围绕串口通信展开，适合有C#基础并希望快速上手上位机开发的工程师。工程通过System.IO.Ports与下位机交互，支持设置波特率、校验位、数据位等串口参数，利用Windows Forms或WPF构建实时刷新的数据显示面板，并将采集结果保存到数据库中，覆盖通信、界面与存储三大环节。压缩包共95个文件、约1.97MB，其中包含28个C#源文件（.cs）、7个可执行程序、7个资源文件（.resx）、6个配置文件及项目解决方案等，另有PDF说明和CSV示例数据，便于对照工程结构理解串口配置、数据绑定与持久化实现。工程还附带2018年7月4日NEDC相关采集记录，可作为车辆测试数据采集场景的参考模板，已有5284人学习下载，适合需要完整示例以快速搭建自己的数据采集系统开发者。

1. 上位机数据采集、储存、实时显示：为什么你的曲线总比现场慢半拍

一台测试台设备每 100ms 上报一组温度和压力数据，上位机界面上的曲线却走走停停，回看历史记录时发现丢了两小时的数据。这种现场我碰过太多次——协议解析本身没难度，难在三个动作的节奏对齐：采集要快，储存要稳，实时显示不能反过来拖垮采集。这篇笔记围绕“上位机数据采集、储存、实时显示”拆一套能直接复用的实现，从通信选型讲到 SQLite 入库，再到 pyqtgraph 实时曲线，最后把半包、串口溢出、写库冲突、时间戳错位这些真坑逐个说清楚。想给非标设备做数据记录和监控的工程师可以直接照着做，熟手重点看第四、五章的参数与踩坑。

2. 架构先行：采集、储存、显示三个独立节奏怎么选型

把“上位机”这个黑匣子掀开，里面永远是三件事在同时跑：从设备拿数据、把数据写进存储、把最新数据画到界面。不少人一上来就写一个 while 循环，读一次串口、存一条库、刷一次图，代码挺短，现场必翻车。原因很简单：串口读取是阻塞的，磁盘写入是有延迟的，界面重绘是耗 CPU 的，三个节奏挤在一起，任何一个卡顿都会让另外两个跟着遭殃。所以架构上要先拆开这三件事，各自用独立的线程和节奏跑。

2.1 通信层选型：串口、网口、Modbus，先定协议再写代码

上位机采集的第一步不是打开 IDE，而是确定设备怎么把数据吐出来。工业现场最常见的通道就三类：RS232/RS485 串口，TCP/UDP 网口，以及在两种物理链路上跑的 Modbus RTU/TCP 这类标准协议。串口适合近距离、点位少的仪表和 PLC，网口适合数据量大、需要跨设备走交换机的场景；Modbus 则是统一了帧格式的标准解法，能对上协议就能省掉大把自定义解析时间。

我一般会先做一次“抓包前置”：用串口调试助手或网络调试工具抓一把原始报文，确认三件事——通信参数是什么、帧头和帧尾是什么、设备每隔多久发一帧。这三个事实直接决定后面的采样设计和缓冲参数。拿串口举例，一台流量计可能跑 115200 波特率、8 数据位、1 停止位、无校验，每 100ms 发一个 ASCII 文本帧；而一块采集板可能是 Modbus RTU 二进制帧，需要先过 CRC 再解析。这个动作没有替代品，能省掉后续 80% 的“数据对不上”类问题。

这里要提前说一句：串口的“帧”由时序决定，驱动层 read 一次未必能拿回完整一帧，这个半包问题第五章专门讲。抓包记录留好，后面写解析器就是对照它来的。

2.2 采集层：为什么必须用“读线程 + 队列”而不是定时器直读

很多第一次做上位机的人会问：界面刷新时顺便读一下串口不行吗？不行。串口 read 是阻塞的，设备没数据时 read 会一直等，超时多久取决于参数设置；而界面每 30ms 要刷一次曲线，两件事天然互相拖累。更麻烦的是，界面一旦卡住，串口驱动里的接收缓冲区会被新数据占满，旧数据被系统直接覆盖丢弃。数据不是没采到，是被界面卡没了。

业内通用做法是生产者-消费者模型：开一个独立采集线程，只负责从串口读字节、解析出完整帧，再把帧放进一个先进先出队列；存储和显示各自按自己的节奏从队列消费数据。队列把“读得快”和“用得慢”解耦，显示端偶尔卡一卡，数据仍然安全地等在队列里。这个结构做中继服务、协议网关也都是同一套思路。

设计队列缓冲时记住一个公式：队列容量要大于“显示或存储最大停顿时间 × 采样频率”。这样即使界面卡住几秒，数据也不会被挤掉。具体数值第 4 章给一张可以直接抄的参数表。

2.3 存储层：点位和频率算清楚，再决定用 SQLite 还是时序库

存储方案常被放到最后考虑，结果数据量一上来就翻车。估算方法很直接：数据量 = 测点点位 × 采样频率 × 单点占用字节 × 运行时长。一台设备 20 个点位、每 100ms 一帧、一天跑 12 小时，一天就是 20 × 10 × 3600 × 12 ≈ 864 万个数值点。这个量级 SQLite 单机文件完全吃得下，写得合理时每秒处理几千行没问题；要是 50 个点位、每 10ms 一帧、常年挂机，就得考虑 MySQL 或时序库这类服务型方案。

我的经验是先按“单机低运维”起步，确认每秒写入超过几千条再升级。SQLite 胜在零部署、单文件备份方便；配合 WAL 日志模式和批量事务，在大多数非标采集场景里比 MySQL 省心得多。时序库的优势在扩展性和聚合查询，但要搭建服务、考虑保留策略，初期成本高出不少。回到这个标题，绝大多数读者场景的默认答案就是 SQLite，服务器型数据库是数据量真上去了再迁移的选项。

2.4 显示层：实时曲线不等于每帧都刷新

实时显示最容易踩的误区是“采样频率多高，界面就多高刷新”。100ms 采一次并不需要每 100ms 重绘一次曲线——绘图本身耗 CPU，重绘越频繁，界面越抢资源，反而拖慢采集线程。常见做法是用一个独立 QTimer，按 30~100ms 的周期去队列里取“这段期间累积的新点”，然后一次性画上去。显示周期略大于采样周期是正常的，实时性足够，界面还稳。

绘图库我推荐 pyqtgraph，比 matplotlib 更适合高频滚动曲线：刷新不闪烁、支持缩放和区域选择，几乎就是为采集监控设计的。下一章直接给代码。还要注意显示和存储不要共用同一个消费队列，否则一条数据被显示拿走，存储就少一条。我的做法是采集线程解析后同时往两个队列各放一份，一个给存储线程，一个给显示定时器。

3. 10分钟跑通最小采集系统：串口半包解析 + SQLite 批量入库 + pyqtgraph 实时曲线

这一章给一套能整段复制改用的 Python 实现，依赖只有四个：pyserial、PySide6、pyqtgraph、Python 自带的 sqlite3 和 queue。手边没有真实设备时，用虚拟串口工具创建一对互连的 COM1/COM2，然后让一段脚本在 COM1 上扮演设备发数据，采集程序读 COM2。下面依次是模拟设备、采集解析、存储入库、实时曲线四段代码。

3.1 模拟设备端：没有真实设备也能先跑通全链路

先用虚拟串口工具把 COM1 和 COM2 对接成一对，往 COM1 写的数据会原样出现在 COM2。然后存一个 device_sim.py，模拟一块每 100ms 上报一次的采集板：

import serial import struct import time import random ser = serial.Serial("COM1", 115200, timeout=0.1) cnt = 0 while True: cnt += 1 value = 1000 + random.randint(-50, 50) # 帧结构：AA 55 序号(2字节) 数值(2字节) 0D 0A frame = b"\xAA\x55" + struct.pack("<Hh", cnt & 0xFFFF, value) + b"\x0D\x0A" ser.write(frame) time.sleep(0.1) # 10Hz，对应每100ms一帧

这个脚本的核心是构造一个带帧头、序号、数值、帧尾的固定二进制帧。帧头 AA 55 用来在字节流里定位起点，序号用来做丢帧统计，数值用小端序打包，帧尾放 0D 0A 方便识别。sleep(0.1) 控制上报频率为 10Hz，这是测试采集链路最常用的频率。

注意：真实设备可能一分钟才一帧，也可能 1kHz 高频。改 sleep 就能改频率，不影响整体框架，只影响第 4 章的缓冲参数。

3.2 采集端：读线程、半包重组与双队列分发

采集端要解决的核心问题是：read 返回的字节数不确定，必须自己做帧重组。下面这段采集线程把解析好的 (时间戳, 序号, 数值) 元组同时放进两个队列，一个是存储队列，一个是显示队列：

import serial import struct import time import queue from threading import Thread store_queue = queue.Queue(maxsize=2000) plot_queue = queue.Queue(maxsize=2000) def read_loop(port="COM2", baud=115200): ser = serial.Serial(port, baud, timeout=0.2) buf = bytearray() while True: chunk = ser.read(128) # 每次最多读128字节 if not chunk: continue buf.extend(chunk) while True: idx = buf.find(b"\xAA\x55") if idx < 0: # 没找到完整帧头，只保留末尾可能被截断的2字节 buf = buf[-2:] if len(buf) > 2 else bytearray() break if idx > 0: del buf[:idx] if len(buf) < 6: # 一帧6字节还没凑齐，等下次read break if buf[4] == 0x0D and buf[5] == 0x0A: cnt, value = struct.unpack("<Hh", bytes(buf[2:4])) ts = time.time() item = (ts, cnt, value) store_queue.put(item) # 交给存储线程 plot_queue.put(item) # 交给显示定时器 del buf[:6] else: del buf[:2] # 帧头后面不是预期帧尾，丢弃伪帧头

这段代码做了两件事：外层不断把 read 到的字节追加到 buf，内层 while 完成“找帧头 → 凑够长度 → 校验帧尾 → 取出数据”的重组循环。timeout=0.2 保证设备空闲时不会无限阻塞，read(128) 控制单次读取量，避免一次读太多导致高水位卡顿。

这里特意设计了两个队列而不共用一个，是因为存储线程和显示定时器消费数据的节奏完全不同：存储希望攒一批写一次，显示希望每 50ms 拿到最新点。共用一个队列会让两边互相抢数据，一人拿一条，最终数据库和曲线各缺一半。双队列多占一点内存，换回的是逻辑清晰。

3.3 存储端：SQLite WAL 与批量入库

存储线程只做一件事：从 store_queue 取数，攒够一批再一次性写入数据库。这样避免每条数据触发一次磁盘事务，吞吐性能能差出一个量级：

import sqlite3 conn = sqlite3.connect("data.db", check_same_thread=False) conn.execute("PRAGMA journal_mode=WAL") conn.execute("""CREATE TABLE IF NOT EXISTS samples ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL, count INT, value REAL )""") conn.execute("CREATE INDEX IF NOT EXISTS idx_ts ON samples(ts)") def save_loop(): batch = [] while True: try: item = store_queue.get(timeout=1) batch.append(item) except queue.Empty: pass if len(batch) >= 100: # 攒够100条一次性写入并提交 conn.executemany( "INSERT INTO samples(ts, count, value) VALUES(?,?,?)", batch) conn.commit() batch.clear()

关键点在三个地方。executemany 比逐条 execute 快很多，这是吞吐的核心；commit 由 batch 积累到阈值触发，而非每条一提交；check_same_thread=False 让同一个连接可以在采集线程之外的其他线程里使用。PRAGMA journal_mode=WAL 的作用是让读和写可以同时进行，后面做历史回放查询时不会被写阻塞。

注意：WAL 模式下目录里会出现 data.db-wal、data.db-shm 两个伴随文件，程序正常退出后会自动合并回主库。日常备份时要把这三个文件一起拷贝，或者先执行一次 CHECKPOINT。

3.4 显示端：pyqtgraph 滚动曲线与刷新定时器

显示端用 QTimer 控制重绘周期，不跟设备上报频率绑定。每次定时器触发时，把 plot_queue 里积攒的新点全部取出，追加到显示数组，再做截断和重绘：

import time import pyqtgraph as pg from PySide6 import QtCore, QtWidgets import sys app = QtWidgets.QApplication(sys.argv) win = pg.GraphicsLayoutWidget(show=True, title="实时曲线") curve = win.addPlot().plot(pen="y") points_x = [] # 相对时间，单位秒 points_y = [] # 物理量数值 start_ts = time.monotonic() MAX_POINTS = 600 # 只保留最近600个点，10Hz下对应60秒窗口 def on_timer(): got = False while True: try: item = plot_queue.get_nowait() except queue.Empty: break points_x.append(time.monotonic() - start_ts) points_y.append(item[2]) # item = (ts, count, value) got = True if len(points_x) > MAX_POINTS: del points_x[: -MAX_POINTS] del points_y[: -MAX_POINTS] if got: curve.setData(points_x, points_y) timer = QtCore.QTimer() timer.timeout.connect(on_timer) timer.start(50) # 每50ms刷新一次，约20帧/秒 app.exec()

这里有两个细节值得注意。X 轴用的是 time.monotonic() 相对时间，而不是系统 time.time()。monotonic 不受系统校时影响，运行中改系统时间曲线不会突然跳一大截。窗口长度 MAX_POINTS 直接控制 setData 的数据量，600 点对 pyqtgraph 来说是极小负载；如果窗口要拉到几小时，就必须对数据做降采样，第 4 章末尾会讲。

3.5 启动顺序与丢帧验证

整个系统按下面顺序启动：先建虚拟串口对，再跑 device_sim.py，最后跑包含采集、存储、显示的主程序。验证分三层：看曲线是否在动、看库表行数是否增长、用帧序号做完整性校验。

帧序号完整性可以直接用 SQL 算。设备从 0 连续递增，理论上最小序号到最大序号之间的总数，减掉表里实际行数，就是链路丢失的帧数：

SELECT COUNT(*) AS actual, MAX(count) - MIN(count) + 1 AS expected, MAX(count) - MIN(count) + 1 - COUNT(*) AS lost FROM samples WHERE ts BETWEEN 1690000000 AND 1690000060;

这个查询在设备不重启的前提下成立。设备重启会使序号跳变，排查时要把重启点单独排除。把这条 SQL 存成脚本，每次现场联调跑一遍，是判断“丢没丢数据”最直接的手段，比肉眼看曲线可靠得多。

4. 必调参数：采样率、队列、批量入库与刷新周期怎么配

代码跑通只是开始，真正决定系统稳不稳的是参数。采样频率、队列容量、批量条数、刷新周期这四个参数互相牵制，配不好就会出现“曲线正常但数据库缺一段”“界面流畅但设备缓冲区溢出”这类隐性故障。下面按四组参数分别讲怎么定。

4.1 采样频率与串口吞吐上限：先算波特率再说别的

串口每秒能传的字节有一个硬上限，工程估算一般按“波特率除以 10”算字节速率。原因是串口传输一个字节实际要发起始位、数据位、停止位共 10 个 bit（带校验要多 1 bit）。那么 115200 波特率大约每秒 11520 字节，也就是约 11.5KB/s。

假设设备每帧 6 字节，理论上 115200 波特率每秒最多传 1920 帧，支撑 1kHz 采样看起来够，但这是理论值。协议解析、系统调度、驱动延迟都会吃掉一部分吞吐，现场要留一半以上余量。

采样间隔必须大于一帧的传输时间，这个判断在频段里占一点：假设你给设备下发“开始采样”指令后设备每 5ms 回一帧，而一帧就要 6ms 传完，那上位机永远在追数据，缓冲区必然溢出。常见做法是把设备上报周期设计成裸传输时间的三倍以上。

4.2 队列与批量入库：一组可以直接抄的参数表

队列容量和批量条数是配套设置的。队列太小，显示或存储稍微卡一下就把队列挤满；队列太大，程序崩溃时队列里还没落盘的数据全丢。批量条数决定事务提交频率，条数越多吞吐越高，但崩溃时丢失的数据窗口也越大。

我常用的起点参数如下表，按采样频率分档，实际按现场最大停顿时间微调：

flush 周期 = 批量条数 ÷ 采样频率。工程上 flush 周期压在 1 到 10 秒之间比较合适，既能获得批量写的高吞吐，崩溃时丢数据也控制在可承受范围内。如果现场对“丢最后 5 秒数据”完全不能忍，把批量条数调小，代价是写库频次上升、总体吞吐下降。

4.3 显示刷新与降采样：高点位时别让 setData 成为瓶颈

显示端的刷新周期和窗口点数也是一组要搭配的参数。刷新太频繁没什么收益，人眼对 30fps 以上的变化已经感知不到，而 CPU 占用和功耗会明显上涨。50ms 刷新（20fps）在绝大多数场景是甜点区间。

窗口点数决定每次 setData 传多少数据。几千点以内 pyqtgraph 都能轻松处理，但如果窗口拉长到几万点，每次重绘全量数据就会卡。我的习惯是超过窗口显示容量后做“降采样”，不是简单隔点抽，而是对一段区间取均值或取峰值。实时监控曲线关心的是趋势和异常跳变，聚合后画出来的曲线反而更接近肉眼该看到的东西。

5. 上位机采集常见问题排查：半包、溢出、锁库与时间戳错位

排查这类问题前先确认两个前提：一是你有没有丢帧统计日志，二是问题能不能稳定复现。能复现就用控制变量法逐步关掉显示、存储，只留采集，看问题是否消失；不能复现就从帧序号统计里找异常时间段，再对照日志定位。下面这五个坑是我在数据采集项目里反复遇到的，每一条都按现象、原因、解决三步拆开。

5.1 半包与粘包：曲线出现毛刺或整段断帧

现象：数据库里某几个点的数值突然跳变，或者设备明明每秒都在发数据，库里却缺了整整几秒。用串口调试助手抓包时数据正常，程序跑起来就不对。

原因：串口 read 不保证一次拿回完整一帧。read 可能在帧中间返回，这是半包；也可能设备连续两帧拼接在一起被 read 一次返回，这是粘包。直接用 read 返回内容做解析，必然丢帧或错位。很多人把 read 当成“读一帧”，这是最隐蔽的判断错误。

解决：用字节缓冲累积 read 结果，按“帧头定位 + 固定帧长 + 帧尾校验”三重规则重组。第 3.2 节的代码就是标准解法。实战中还有一个容易漏掉的点：帧头 AA 55 可能出现在数据负载里，所以只找帧头不够，还必须校验帧尾 0D 0A，校验不过就要丢弃当前伪帧头并继续搜索下一个 AA 55。

5.2 串口缓冲区溢出：只要程序一卡，数据就丢一段

现象：程序平时运行正常，但界面拖动窗口、CPU 被其他任务占满时，数据库里就会出现一段连续缺失。等程序恢复，设备还在继续发，缺的那段再也补不回来。

原因：操作系统串口接收缓冲区是有限的，通常只有几 KB。采集线程一旦没来得及 read，驱动缓冲区被填满，新来的数据直接丢弃。这不是程序逻辑错，是“读得太慢”导致的物理丢数据。

解决：把采集线程的 read 放到独立线程并提高消费速度；read 时用 timeout 防止无限阻塞；界面显示与采集彻底解耦，显示卡顿不应阻塞 read。再配合 4.2 节的队列参数，给足够缓冲余量。现场实在要高频大流量，可优先换 USB 转串口芯片以及缩短线缆，驱动丢数据概率会低一些。

5.3 界面卡顿与曲线闪烁

现象：曲线显示正常，但窗口拖动、最小化再恢复时明显卡顿；或者曲线重绘时伴随闪烁。

原因：常见两种。一是绘图的 setData 与采集处理在同一个线程，采集解析阻塞了绘图；二是每次刷新传入上万点，绘图库处理不过来。pyqtgraph 的抗闪烁能力比 matplotlib 强很多，但数据量过大照样卡。

解决：显示定时器独立运行，不要直接在采集线程里调 setData；控制窗口点数，超过容量就截断或降采样；对 start 时间戳用单调时钟而不是系统时间，避免系统校时导致曲线瞬间跳变。卡顿复现时先关掉存储只看显示，再关掉显示只看存储，能快速定位是哪一层拖了后腿。

5.4 SQLite 报错 database is locked

现象：程序运行一段时间后控制台或日志里出现 database is locked，严重的连 SELECT 查询也报错，必须重启程序才能恢复。

原因：SQLite 在多线程并发写时默认行为是锁库。一个连接在写，另一个连接也在写，后者会拿不到写锁直接报错。最常见场景是“显示端为了回放历史也开了个连接”，和存储线程在同一时刻写库。默认 journal 模式下读请求也会被写事务阻塞。

解决：开库后先执行 PRAGMA journal_mode=WAL，让读写并发；写操作统一走存储线程，显示回放连接只做 SELECT。如果多个模块都要写采样数据，最好加一层应用层队列，把所有写请求排到一个线程处理。单写通道是 SQLite 这类嵌入式数据库的铁律。

5.5 时间戳错位：回放对不上真实发生时刻

现象：历史曲线和设备原始记录对不上，峰谷明显错位；或者设备重启后，当前曲线出现倒走一段再继续前进的诡异形状。

原因：绝大多数是时间戳不是在“数据产生时刻”盖的。比如在 read 返回后立刻取 time.time()，但此时拿到的是多个帧的混合块，重组后已经过了几十毫秒；更常见的是 X 轴用了设备自己的计数器，设备一重启计数器归零，曲线自然倒走。

解决：时间戳必须在“完整帧已经校验通过”的那个瞬间取，也就是 3.2 节代码里 del buf[:6] 之前那一行。显示端 X 轴优先用上位机时间戳，并尽量用单调时钟换算成相对时间。如果设备自身带时标字段，则以上位机收到时刻为基准、设备时标作为辅助参考。排查阶段把“上位机时间”和“设备序号”同时打印出来，对比异常区间一眼就能看出谁的问题。

6. 进阶验证：丢帧统计、离线回补与断电恢复

把最小系统跑稳定之后，下一步是把系统从“能用”做成“可查”。首推的是丢帧统计常态化：在采集程序里维护设备序号，每次解析帧时检查 count 是否连续，不连续就写入一个丢帧日志表，记录丢帧起止序号和时间戳。这样即使没有现场盯着，事后也能知道哪段时间链路出了问题。对应 SQL 就是 3.5 节那条查询，不过要把“重启导致序号跳变”的情况单独排除，否则会把正常重启误报成丢帧。

离线回补是丢帧后的补救手段。如果设备协议支持按序号重发历史数据，采集程序发现序号不连续时，可以下发“从某序号开始重发”的命令，把缺失区间补回来。补回来的数据要带一个标记字段，比如 is_recovered，避免后续统计时把补数当成实时数据影响告警判断。不支持回补的设备，只能靠丢帧日志做记录留证。

断电恢复这件事要在设计时就留位置。SQLite 的 WAL 模式本身带崩溃恢复能力，程序异常退出后，下次启动打开库时 WAL 会自动重放。但要提醒一句：第 4.2 节批量提交的窗口期内未 commit 的数据，断电后大概率丢失，这个窗口大小就是你接受的丢数时长。重要场景下建议把 batch 条数调小，或者加一个独立的 CSV 补记文件作为兜底，采集线程每解析一帧就顺手 append 一行，两条链路独立落盘，故障时间点用 CSV 反推最可靠。

回放进阶功能也可以捣鼓一下：给显示界面加一个“加载历史”按钮，从数据库按时间窗口取数重绘成曲线。这里注意两点，一是历史查询要走单独只读连接，避免抢写锁；二是历史数据可能几万个点，查询后先做聚合再 setData，不要一股脑传给绘图控件。

数据采集这块做到最后，你会发现代码本身并不难，难的是把采集、储存、显示当成一整条链路，并且把每一次异常都变成以后可以查的记录。我早期有一版采集程序没做丢帧统计，现场偶发跳变让我排查了一星期，后来加上了帧序号校验和完整性检查，同类问题半小时就能定位。从那以后，我写任何采集系统都会把“采集线程、双队列、批量落盘、丢帧日志”这四个动作先固定成模板，再来填具体协议。希望这一篇能让你跳过我踩过的这些坑，直接落到一个稳定可查的采集底座上。

本文还有配套的精品资源，点击获取