1. 串口数据采集这件事远比想象中琐碎串口通信在工业控制、传感器采集、嵌入式调试这些场景里几乎是绕不开的基础能力。很多人第一次在Linux下接串口设备以为打开/dev/ttyUSB0读数据就完事了结果上手才发现数据断帧、乱码、丢包、权限报错、拔插之后设备名变了、程序跑着跑着CPU飙到100%……这些问题一个接一个。标题里说的读取、处理、保存三个动作看起来简单实际上每一个都有不少门道。这篇文章面向的是需要在Linux环境下做串口数据采集的开发者不管你是刚接触串口的新手还是已经写过几版采集程序但总觉得不够稳的老手都能从中找到可以直接用的东西。我会从设备识别、串口参数配置、数据读取策略、协议解析、落盘存储、异常处理这几个维度把整个链路拆开讲清楚并且给出可以直接编译运行的C代码示例。选C是因为串口采集通常跑在资源受限的嵌入式Linux或者工控机上C的可控性最好当然Python方案我也会提一下适用场景。核心关键词就三个串口读取、数据处理、数据保存。但要把这三个词落地涉及的知识面其实挺宽的——termios配置、非阻塞IO、select/poll多路复用、环形缓冲区、帧同步、文件IO缓冲策略、日志轮转等等。下面我按实际开发中会遇到的顺序一层一层往下讲。2. 先搞清楚你面对的是哪个串口设备2.1 设备节点是怎么冒出来的Linux下串口设备通常以/dev/ttyS*板载串口、/dev/ttyUSB*USB转串口、/dev/ttyACM*USB CDC设备的形式出现。当你插上一个CH340、CP2102或者FT232芯片的USB转串口模块内核的usbserial驱动会加载然后在/dev下创建对应的设备节点。用dmesg | tail就能看到内核打印出来的信息比如识别到ch341-uart转换器 attached to ttyUSB0。这里第一个坑就来了设备名不固定。你今天插上模块是ttyUSB0明天换个USB口可能就变成ttyUSB1同时插两个模块更是分不清谁是谁。如果程序里硬编码/dev/ttyUSB0换个环境就跑不起来。解决办法有两个方向。一是用udev规则根据设备的VID/PID或者序列号创建固定名称的符号链接比如在/etc/udev/rules.d/下写一条规则把特定厂商的串口设备映射成/dev/sensor_a这样的固定名字。二是程序启动时扫描/dev/serial/by-id/目录这个目录下的链接名包含了设备序列号天然唯一。我一般推荐第二种因为不需要动系统配置程序自己就能搞定。# 查看串口设备的详细信息 ls -l /dev/serial/by-id/ # 输出示例 # usb-1a86_USB_Serial-if00-port0 - ../../ttyUSB02.2 权限问题为什么open总是返回Permission denied普通用户默认没有权限访问/dev/ttyUSB*因为设备节点的属主是root属组是dialout。你有两个选择把当前用户加入dialout组sudo usermod -aG dialout $USER然后重新登录生效或者用sudo跑程序。前者是正道后者在开发阶段图省事可以用但生产环境绝对不要用root跑采集程序一旦程序有漏洞后果不堪设想。还有一个容易被忽略的点串口被占用。如果有一个进程已经打开了这个串口你再open会返回EBUSY。排查方法是fuser /dev/ttyUSB0或者lsof | grep ttyUSB看看是谁占着。有时候是上次的程序没退干净有时候是ModemManager这个系统服务在自动探测串口设备它会短暂打开串口发AT指令导致你的程序打不开。如果确认不需要ModemManager管理你的设备可以把它禁用掉或者用udev规则让ModemManager忽略特定设备。2.3 串口参数不只是波特率那么简单打开串口之后必须用termios结构体配置通信参数。很多人只知道设波特率其实数据位、停止位、校验位、流控每一个都可能出问题。参数常见值说明波特率9600 / 115200 / 460800必须和对方设备一致不一致就是乱码数据位8绝大多数设备用8位停止位1少数设备用2位校验位None / Even / Odd工业设备常用Even校验流控None / RTSCTS高速通信建议开硬件流控配置的核心代码大概长这样#include termios.h #include fcntl.h #include unistd.h int open_serial(const char *dev, int baud) { int fd open(dev, O_RDWR | O_NOCTTY | O_NONBLOCK); if (fd 0) { perror(open serial); return -1; } struct termios tty; if (tcgetattr(fd, tty) ! 0) { perror(tcgetattr); close(fd); return -1; } cfsetospeed(tty, baud); cfsetispeed(tty, baud); tty.c_cflag | (CLOCAL | CREAD); // 忽略调制解调器控制线启用接收 tty.c_cflag ~CSIZE; tty.c_cflag | CS8; // 8位数据位 tty.c_cflag ~PARENB; // 无校验 tty.c_cflag ~CSTOPB; // 1位停止位 tty.c_cflag ~CRTSCTS; // 无硬件流控 tty.c_lflag ~(ICANON | ECHO | ECHOE | ISIG); // 原始模式 tty.c_iflag ~(IXON | IXOFF | IXANY); // 关闭软件流控 tty.c_iflag ~(INLCR | ICRNL | IGNCR); // 不做换行转换 tty.c_oflag ~OPOST; // 原始输出 tty.c_cc[VMIN] 0; // 非阻塞读 tty.c_cc[VTIME] 10; // 读超时1秒单位是0.1秒 if (tcsetattr(fd, TCSANOW, tty) ! 0) { perror(tcsetattr); close(fd); return -1; } tcflush(fd, TCIOFLUSH); // 清空收发缓冲区 return fd; }这里有几个细节值得展开说。O_NOCTTY是告诉内核这个串口不要成为控制终端否则程序可能会收到SIGINT之类的信号。O_NONBLOCK配合VMIN0和VTIME实现超时读这样read不会永久阻塞。tcflush在打开后清一次缓冲区很重要因为串口里可能残留着上次通信的垃圾数据。注意VMIN和VTIME的组合行为容易搞混。VMIN0, VTIME0表示read最多等VTIME个0.1秒有数据就返回数据没数据就返回0。VMIN0, VTIME0表示read至少等到VMIN个字节才返回否则一直阻塞。VMIN0, VTIME0表示第一个字节到达后开始计时超时或凑够VMIN字节就返回。实际采集场景我一般用第一种配合select做超时控制。3. 读数据的方式选不对后面全是坑3.1 阻塞读、非阻塞读、多路复用怎么选串口读取有三种基本模式。阻塞读最简单read没数据就卡在那里但问题是程序没法同时干别的事也没法优雅退出。非阻塞读配合轮询CPU空转浪费资源。多路复用select/poll/epoll是实际项目里最常用的既能等数据又能设超时还能同时监听多个文件描述符。我个人的选择是单串口采集用poll就够了代码比select清爽没有FD_SETSIZE的限制。如果要同时采集多个串口epoll在大量描述符时性能更好但串口数量一般不会太多poll完全够用。#include poll.h int read_with_timeout(int fd, unsigned char *buf, int maxlen, int timeout_ms) { struct pollfd pfd; pfd.fd fd; pfd.events POLLIN; int ret poll(pfd, 1, timeout_ms); if (ret 0) { perror(poll); return -1; } if (ret 0) { return 0; // 超时无数据 } if (pfd.revents POLLIN) { return read(fd, buf, maxlen); } return -1; }3.2 数据断帧为什么你收到的数据总是缺胳膊少腿串口是字节流没有消息边界的概念。对方设备发一帧20字节的数据你的read可能第一次返回8字节第二次返回12字节也可能两次数据粘在一起返回30字节。这不是bug是串口通信的本质。必须自己实现帧同步逻辑。常见的帧格式有两种定长帧和变长帧。定长帧好办攒够N字节就处理。变长帧通常有帧头、长度字段、数据区、校验字段比如0xAA 0x55 | len | data... | crc。解析的时候需要一个状态机逐字节扫描找到帧头后根据长度字段确定整帧长度凑齐后校验校验通过才交给业务层。typedef enum { STATE_IDLE, STATE_HEADER1, STATE_HEADER2, STATE_LENGTH, STATE_PAYLOAD, STATE_CHECKSUM } parse_state_t; typedef struct { parse_state_t state; unsigned char buf[512]; int index; int expected_len; } frame_parser_t; void parser_init(frame_parser_t *p) { p-state STATE_IDLE; p-index 0; p-expected_len 0; } // 返回1表示解析出一帧完整数据0表示还在解析中 int parser_feed(frame_parser_t *p, unsigned char byte) { switch (p-state) { case STATE_IDLE: if (byte 0xAA) { p-buf[0] byte; p-index 1; p-state STATE_HEADER2; } break; case STATE_HEADER2: if (byte 0x55) { p-buf[1] byte; p-index 2; p-state STATE_LENGTH; } else { p-state STATE_IDLE; } break; case STATE_LENGTH: p-buf[2] byte; p-expected_len byte 4; // 帧头2 长度1 数据len 校验1 p-index 3; p-state STATE_PAYLOAD; break; case STATE_PAYLOAD: p-buf[p-index] byte; if (p-index p-expected_len - 1) { p-state STATE_CHECKSUM; } break; case STATE_CHECKSUM: p-buf[p-index] byte; // 这里做校验假设是简单的累加和 { unsigned char sum 0; for (int i 0; i p-index - 1; i) { sum p-buf[i]; } if (sum byte) { p-state STATE_IDLE; return 1; // 完整帧 } } p-state STATE_IDLE; break; } return 0; }这个状态机的思路是逐字节驱动每收到一个字节就推进状态。好处是不管数据怎么分片到达解析结果都一样。实际项目中我会把read到的数据先塞进一个环形缓冲区然后解析器从环形缓冲区里取字节这样读写解耦逻辑更清晰。3.3 环形缓冲区串口采集的标配组件为什么需要环形缓冲区因为read返回的数据可能包含多帧也可能只有半帧。如果直接在read的缓冲区上做解析处理起来很别扭。环形缓冲区ring buffer提供先进先出的字节流抽象写入端只管往里塞读取端只管从里取两边互不干扰。#define RING_SIZE 4096 typedef struct { unsigned char data[RING_SIZE]; volatile int head; // 写入位置 volatile int tail; // 读取位置 } ring_buf_t; int ring_write(ring_buf_t *r, const unsigned char *src, int len) { int available (r-tail - r-head - 1 RING_SIZE) % RING_SIZE; if (len available) len available; for (int i 0; i len; i) { r-data[r-head] src[i]; r-head (r-head 1) % RING_SIZE; } return len; } int ring_read(ring_buf_t *r, unsigned char *dst, int len) { int used (r-head - r-tail RING_SIZE) % RING_SIZE; if (len used) len used; for (int i 0; i len; i) { dst[i] r-data[r-tail]; r-tail (r-tail 1) % RING_SIZE; } return len; }如果采集线程和解析线程是两个线程head和tail的读写需要考虑内存屏障或者用原子操作。单线程场景下用volatile就够了。这个环形缓冲区的大小要根据数据速率来定115200波特率下每秒最多约11KB数据4KB缓冲区大概能缓冲0.36秒一般够用。如果解析逻辑比较重可以适当加大。4. 数据处理从原始字节到可用信息4.1 字节序和数据类型转换串口传过来的多字节数据字节序大端/小端取决于发送方设备。很多传感器用大端序网络字节序而x86 Linux是小端直接memcpy到int会得到错误的值。必须手动组装// 大端序4字节转uint32 uint32_t be32_to_u32(const unsigned char *p) { return ((uint32_t)p[0] 24) | ((uint32_t)p[1] 16) | ((uint32_t)p[2] 8) | ((uint32_t)p[3]); } // 小端序4字节转uint32 uint32_t le32_to_u32(const unsigned char *p) { return ((uint32_t)p[3] 24) | ((uint32_t)p[2] 16) | ((uint32_t)p[1] 8) | ((uint32_t)p[0]); }浮点数的处理更麻烦有些设备用IEEE 754格式直接传有些用定点数。如果是IEEE 754先按字节序组装成uint32_t再用memcpy转成float。如果是定点数需要知道小数位数比如温度值放大100倍传输收到之后除以100.0。4.2 物理量换算与校准原始ADC值往往需要换算成物理量。比如一个12位ADC参考电压3.3V那么电压值 raw / 4095.0 * 3.3。如果是电流环4-20mA的传感器还要做线性映射。这些换算公式通常写在设备手册里但实际使用中往往需要现场校准——因为传感器有零漂和增益误差。我一般会在程序里留一组校准参数支持通过配置文件或者命令行参数调整。比如温度传感器实测发现比标准温度计低0.5度那就加一个偏移量。这种校准值不要硬编码在代码里否则换个传感器就得重新编译。4.3 数据滤波什么时候该滤什么时候不该滤采集到的数据往往有噪声常见的滤波方法有滑动平均、中值滤波、一阶低通滤波。滑动平均最简单维护一个固定长度的队列每次新数据进来替换最老的输出平均值。中值滤波对脉冲噪声效果好但需要排序计算量稍大。一阶低通滤波适合变化缓慢的信号公式是y alpha * x (1 - alpha) * yalpha越小越平滑但响应越慢。注意滤波会引入延迟对于需要快速响应的控制场景滤波参数要谨慎选择。另外如果原始数据本身就是要保存的建议先保存原始值再保存滤波值这样后期分析时还有原始数据可用。4.4 时间戳别等到保存的时候才想起来每条数据都应该带时间戳而且时间戳应该在数据到达时就打上而不是在保存时。因为从读取到保存之间可能有排队延迟如果保存时才打时间戳时间就不准了。用clock_gettime(CLOCK_MONOTONIC)获取单调递增时间避免系统时间被NTP调整导致时间戳跳变。如果需要和真实时间对应可以在程序启动时记录一次CLOCK_REALTIME和CLOCK_MONOTONIC的差值之后用单调时间推算真实时间。#include time.h struct timespec ts; clock_gettime(CLOCK_MONOTONIC, ts); long long timestamp_ms (long long)ts.tv_sec * 1000 ts.tv_nsec / 1000000;5. 数据保存写文件也有讲究5.1 文本格式还是二进制格式保存格式的选择取决于后续怎么用。如果数据要给人看或者导入Excel分析CSV文本格式最方便。如果数据量大、写入频繁二进制格式更省空间、写入更快。我一般会同时保存两份一份二进制原始数据用于归档和回放一份CSV摘要用于快速查看。CSV格式的坑在于字段里可能有逗号或换行需要做转义。另外浮点数格式化时要注意精度%.6f通常够用但如果是高精度传感器可能需要更多位数。每行末尾用\n还是\r\n取决于目标平台Linux下用\n就行。二进制格式建议自己定义一个简单的文件头记录魔数、版本号、数据记录的结构体大小、采集开始时间等信息方便以后解析。每条记录定长这样可以用fseek快速定位到第N条记录。5.2 缓冲策略fflush的时机很关键标准IO库fopen/fwrite默认是全缓冲的数据先写到用户空间缓冲区缓冲区满了才真正写到磁盘。如果程序崩溃或者断电缓冲区里的数据就丢了。对于采集程序这是不能接受的。解决方案是定期fflush比如每写100条记录或者每隔1秒调用一次fflush。更严格的做法是用O_SYNC标志打开文件每次write都同步到磁盘但这样性能会差很多。折中方案是用fsync定期同步比如每10秒一次。具体间隔取决于你能容忍多少数据丢失。FILE *fp fopen(data.csv, a); // ... 写入数据 ... if (write_count % 100 0) { fflush(fp); }5.3 文件轮转别让单个文件无限增长长时间运行的采集程序如果一直往一个文件里写文件会越来越大最后可能撑爆磁盘。必须做文件轮转。策略有两种按大小轮转和按时间轮转。按大小就是文件超过比如100MB就关闭当前文件新建一个带序号的文件。按时间就是每小时或每天一个文件文件名里带日期时间。char filename[256]; time_t now time(NULL); struct tm *tm_info localtime(now); strftime(filename, sizeof(filename), data_%Y%m%d_%H%M%S.csv, tm_info);轮转的时候要注意关闭旧文件前一定要fflush确保数据落盘。另外如果磁盘空间有限还需要实现旧文件清理逻辑比如只保留最近7天的数据更早的自动删除。删除前最好确认一下文件已经成功写入避免误删。5.4 断电保护采集程序必须考虑的现实问题工控现场断电是家常便饭。如果程序正在写文件时断电文件可能损坏尤其是CSV这种文本格式最后一行可能是半截的。二进制格式相对好一点但最后一条记录也可能不完整。我的做法是每条记录写入前先写一个长度字段或者记录头解析时如果发现记录头不完整或者长度不对就丢弃最后这条不完整记录。另外文件打开用追加模式a这样即使上次异常退出下次启动还能接着写不会覆盖已有数据。如果对数据完整性要求极高可以考虑先写临时文件定期rename成正式文件rename是原子操作能保证文件要么是完整的旧版本要么是完整的新版本。6. 完整代码框架与编译运行6.1 主循环结构把上面的模块串起来主循环大概是这样#include stdio.h #include stdlib.h #include string.h #include signal.h #include poll.h #include time.h #include termios.h #include fcntl.h #include unistd.h static volatile int running 1; void signal_handler(int sig) { (void)sig; running 0; } int main(int argc, char *argv[]) { const char *dev argc 1 ? argv[1] : /dev/ttyUSB0; int fd open_serial(dev, B115200); if (fd 0) return 1; signal(SIGINT, signal_handler); signal(SIGTERM, signal_handler); ring_buf_t ring; memset(ring, 0, sizeof(ring)); frame_parser_t parser; parser_init(parser); FILE *fp fopen(sensor_data.csv, a); if (!fp) { perror(fopen); close(fd); return 1; } unsigned char read_buf[256]; unsigned char byte; int write_count 0; while (running) { int n read_with_timeout(fd, read_buf, sizeof(read_buf), 100); if (n 0) { ring_write(ring, read_buf, n); } // 从环形缓冲区取字节喂给解析器 while (ring_read(ring, byte, 1) 1) { if (parser_feed(parser, byte)) { // 解析出一帧完整数据 // 假设数据区从buf[3]开始长度为expected_len-4 int data_len parser.expected_len - 4; // 这里做业务处理比如解析出温度值 // 然后保存 struct timespec ts; clock_gettime(CLOCK_MONOTONIC, ts); long long ts_ms (long long)ts.tv_sec * 1000 ts.tv_nsec / 1000000; fprintf(fp, %lld,, ts_ms); for (int i 0; i data_len; i) { fprintf(fp, %02X, parser.buf[3 i]); } fprintf(fp, \n); if (write_count % 100 0) { fflush(fp); } } } } fflush(fp); fclose(fp); close(fd); return 0; }编译命令gcc -o serial_collector serial_collector.c -Wall -O2运行./serial_collector /dev/ttyUSB06.2 Python方案什么时候用什么时候不用如果采集频率不高比如每秒几条数据或者只是做原型验证Python的pyserial库非常方便import serial import time import csv ser serial.Serial(/dev/ttyUSB0, 115200, timeout1) with open(data.csv, a, newline) as f: writer csv.writer(f) while True: data ser.read(ser.in_waiting or 1) if data: ts time.monotonic() writer.writerow([ts, data.hex()]) f.flush()Python的缺点是GIL导致多线程性能受限高频采集时可能丢数据。另外Python进程的内存占用和启动时间也比C大。所以我的建议是原型用Python生产用C。如果团队全是Python背景用Python也可以但要注意用serial.read配合合适的超时避免CPU空转。6.3 调试技巧没有硬件怎么测开发阶段不一定手头有真实设备可以用虚拟串口对来模拟。Linux下用socat可以创建一对互相连接的伪终端socat -d -d pty,raw,echo0 pty,raw,echo0它会输出两个设备路径比如/dev/pts/3和/dev/pts/4。你的程序打开其中一个用另一个发数据就能模拟串口通信。发数据可以用echo -e或者写个小脚本。另一个技巧是用strace跟踪系统调用看看read到底返回了什么open是否成功ioctl有没有报错。strace -e traceread,write,open,ioctl ./serial_collector能快速定位问题。7. 那些只有踩过才知道的坑7.1 USB转串口的热插拔问题USB转串口模块拔掉再插上/dev/ttyUSB0可能变成/dev/ttyUSB1原来的文件描述符会失效read返回-1并且errno是ENODEV。程序必须处理这种情况检测到ENODEV或者EIO时关闭旧fd重新扫描设备尝试重新打开。如果设备名是固定的通过udev规则重新打开同一个路径就行。如果不是需要重新扫描/dev/serial/by-id/。重连逻辑要加退避不能一失败就疯狂重试否则CPU和日志都会爆炸。我一般用指数退避第一次等1秒第二次2秒最多等30秒。7.2 数据量太大导致磁盘写满这个坑很隐蔽程序跑得好好的突然某天发现不写数据了一查磁盘满了。除了文件轮转和旧文件清理还应该监控磁盘剩余空间低于阈值时告警或者暂停采集。可以用statvfs获取磁盘信息#include sys/statvfs.h struct statvfs vfs; statvfs(/data, vfs); unsigned long free_mb vfs.f_bavail * vfs.f_frsize / (1024 * 1024); if (free_mb 100) { // 告警或停止写入 }7.3 串口参数被意外修改有些设备在上电或者复位时会发送一些控制字符如果串口配置了ICRNL把\r转成\n或者IXON软件流控这些字符可能被解释成控制命令导致串口行为异常。所以配置串口时一定要把所有不必要的输入处理关掉用原始模式raw mode。我在代码里把c_iflag和c_lflag里能关的都关了就是为了避免这种意外。7.4 多线程下的资源竞争如果采集和保存分在两个线程共享的环形缓冲区需要加锁或者用无锁队列。加锁简单但可能影响性能无锁队列性能好但实现复杂容易出错。我的经验是如果采集频率在每秒几千条以内用互斥锁完全够用不要过早优化。真正需要无锁队列的场景通常是每秒几十万条以上的高速采集那种场景一般也不会用串口。7.5 日志和数据的分离程序运行日志比如打开串口成功、解析错误和数据文件要分开保存。日志用syslog或者单独的日志文件数据用专门的数据文件。不要把日志和数据混在一个文件里否则后期解析数据时还要过滤日志行很麻烦。日志级别也要控制正常运行时只记warning以上调试时才开debug否则日志文件增长太快。8. 性能调优的几个实际手段8.1 减少系统调用次数每次read都是一次系统调用有开销。如果数据速率很高可以一次read尽量多的数据。VMIN和VTIME的设置会影响read的行为配合poll使用可以在有数据时批量读取。另外read的缓冲区不要太小256字节到4KB之间比较合适。8.2 文件IO的批量写入前面提到用fprintf逐条写每次fprintf都有格式化开销。如果性能要求高可以自己拼字符串到缓冲区攒够一批再write。或者用writev一次性写入多个缓冲区。对于二进制格式直接fwrite结构体数组比逐条写快得多。8.3 CPU占用率优化如果发现程序CPU占用率高先检查是不是在忙等待。poll的超时时间不要设0设0会立即返回变成轮询。设100ms到1s比较合理。另外解析器逐字节处理在大数据量时可能成为瓶颈可以优化成批量扫描帧头找到帧头后再逐字节解析。8.4 内存占用控制环形缓冲区、解析缓冲区、文件写缓冲区的大小都要合理设置。嵌入式设备内存有限不要动不动就分配几MB的缓冲区。根据实际数据速率计算所需的最小缓冲区大小留2-3倍余量即可。9. 从采集到可用数据的完整链路回顾整条链路走下来核心就三件事稳定地读、正确地解析、可靠地存。读的时候注意设备识别、权限、参数配置、超时控制解析的时候注意帧同步、字节序、校验、时间戳存的时候注意格式选择、缓冲策略、文件轮转、断电保护。每一个环节都有坑但每一个坑都有成熟的解决方案。我在实际项目里最深的体会是串口采集程序的健壮性80%取决于异常处理做得好不好。正常流程谁都能写但设备拔插、磁盘满、数据异常、程序被杀这些情况处理不好程序就没法长期稳定运行。所以写这类程序一定要把出错了怎么办想清楚而不是只想着正常情况怎么跑。另外测试阶段一定要用真实设备跑足够长的时间至少24小时以上观察内存有没有泄漏、文件有没有正常轮转、时间戳有没有跳变。很多问题只有在长时间运行后才会暴露出来。如果条件允许模拟断电、拔插串口、填满磁盘这些异常场景验证程序的恢复能力。这些测试做完程序才算真正可用。 
长河君 · 2026/10/15 21:51:41 · 阅读 1.2k Linux串口数据采集实战：读取、处理与保存的完整链路
1. 串口数据采集这件事远比想象中琐碎串口通信在工业控制、传感器采集、嵌入式调试这些场景里几乎是绕不开的基础能力。很多人第一次在Linux下接串口设备以为打开/dev/ttyUSB0读数据就完事了结果上手才发现数据断帧、乱码、丢包、权限报错、拔插之后设备名变了、程序跑着跑着CPU飙到100%……这些问题一个接一个。标题里说的读取、处理、保存三个动作看起来简单实际上每一个都有不少门道。这篇文章面向的是需要在Linux环境下做串口数据采集的开发者不管你是刚接触串口的新手还是已经写过几版采集程序但总觉得不够稳的老手都能从中找到可以直接用的东西。我会从设备识别、串口参数配置、数据读取策略、协议解析、落盘存储、异常处理这几个维度把整个链路拆开讲清楚并且给出可以直接编译运行的C代码示例。选C是因为串口采集通常跑在资源受限的嵌入式Linux或者工控机上C的可控性最好当然Python方案我也会提一下适用场景。核心关键词就三个串口读取、数据处理、数据保存。但要把这三个词落地涉及的知识面其实挺宽的——termios配置、非阻塞IO、select/poll多路复用、环形缓冲区、帧同步、文件IO缓冲策略、日志轮转等等。下面我按实际开发中会遇到的顺序一层一层往下讲。2. 先搞清楚你面对的是哪个串口设备2.1 设备节点是怎么冒出来的Linux下串口设备通常以/dev/ttyS*板载串口、/dev/ttyUSB*USB转串口、/dev/ttyACM*USB CDC设备的形式出现。当你插上一个CH340、CP2102或者FT232芯片的USB转串口模块内核的usbserial驱动会加载然后在/dev下创建对应的设备节点。用dmesg | tail就能看到内核打印出来的信息比如识别到ch341-uart转换器 attached to ttyUSB0。这里第一个坑就来了设备名不固定。你今天插上模块是ttyUSB0明天换个USB口可能就变成ttyUSB1同时插两个模块更是分不清谁是谁。如果程序里硬编码/dev/ttyUSB0换个环境就跑不起来。解决办法有两个方向。一是用udev规则根据设备的VID/PID或者序列号创建固定名称的符号链接比如在/etc/udev/rules.d/下写一条规则把特定厂商的串口设备映射成/dev/sensor_a这样的固定名字。二是程序启动时扫描/dev/serial/by-id/目录这个目录下的链接名包含了设备序列号天然唯一。我一般推荐第二种因为不需要动系统配置程序自己就能搞定。# 查看串口设备的详细信息 ls -l /dev/serial/by-id/ # 输出示例 # usb-1a86_USB_Serial-if00-port0 - ../../ttyUSB02.2 权限问题为什么open总是返回Permission denied普通用户默认没有权限访问/dev/ttyUSB*因为设备节点的属主是root属组是dialout。你有两个选择把当前用户加入dialout组sudo usermod -aG dialout $USER然后重新登录生效或者用sudo跑程序。前者是正道后者在开发阶段图省事可以用但生产环境绝对不要用root跑采集程序一旦程序有漏洞后果不堪设想。还有一个容易被忽略的点串口被占用。如果有一个进程已经打开了这个串口你再open会返回EBUSY。排查方法是fuser /dev/ttyUSB0或者lsof | grep ttyUSB看看是谁占着。有时候是上次的程序没退干净有时候是ModemManager这个系统服务在自动探测串口设备它会短暂打开串口发AT指令导致你的程序打不开。如果确认不需要ModemManager管理你的设备可以把它禁用掉或者用udev规则让ModemManager忽略特定设备。2.3 串口参数不只是波特率那么简单打开串口之后必须用termios结构体配置通信参数。很多人只知道设波特率其实数据位、停止位、校验位、流控每一个都可能出问题。参数常见值说明波特率9600 / 115200 / 460800必须和对方设备一致不一致就是乱码数据位8绝大多数设备用8位停止位1少数设备用2位校验位None / Even / Odd工业设备常用Even校验流控None / RTSCTS高速通信建议开硬件流控配置的核心代码大概长这样#include termios.h #include fcntl.h #include unistd.h int open_serial(const char *dev, int baud) { int fd open(dev, O_RDWR | O_NOCTTY | O_NONBLOCK); if (fd 0) { perror(open serial); return -1; } struct termios tty; if (tcgetattr(fd, tty) ! 0) { perror(tcgetattr); close(fd); return -1; } cfsetospeed(tty, baud); cfsetispeed(tty, baud); tty.c_cflag | (CLOCAL | CREAD); // 忽略调制解调器控制线启用接收 tty.c_cflag ~CSIZE; tty.c_cflag | CS8; // 8位数据位 tty.c_cflag ~PARENB; // 无校验 tty.c_cflag ~CSTOPB; // 1位停止位 tty.c_cflag ~CRTSCTS; // 无硬件流控 tty.c_lflag ~(ICANON | ECHO | ECHOE | ISIG); // 原始模式 tty.c_iflag ~(IXON | IXOFF | IXANY); // 关闭软件流控 tty.c_iflag ~(INLCR | ICRNL | IGNCR); // 不做换行转换 tty.c_oflag ~OPOST; // 原始输出 tty.c_cc[VMIN] 0; // 非阻塞读 tty.c_cc[VTIME] 10; // 读超时1秒单位是0.1秒 if (tcsetattr(fd, TCSANOW, tty) ! 0) { perror(tcsetattr); close(fd); return -1; } tcflush(fd, TCIOFLUSH); // 清空收发缓冲区 return fd; }这里有几个细节值得展开说。O_NOCTTY是告诉内核这个串口不要成为控制终端否则程序可能会收到SIGINT之类的信号。O_NONBLOCK配合VMIN0和VTIME实现超时读这样read不会永久阻塞。tcflush在打开后清一次缓冲区很重要因为串口里可能残留着上次通信的垃圾数据。注意VMIN和VTIME的组合行为容易搞混。VMIN0, VTIME0表示read最多等VTIME个0.1秒有数据就返回数据没数据就返回0。VMIN0, VTIME0表示read至少等到VMIN个字节才返回否则一直阻塞。VMIN0, VTIME0表示第一个字节到达后开始计时超时或凑够VMIN字节就返回。实际采集场景我一般用第一种配合select做超时控制。3. 读数据的方式选不对后面全是坑3.1 阻塞读、非阻塞读、多路复用怎么选串口读取有三种基本模式。阻塞读最简单read没数据就卡在那里但问题是程序没法同时干别的事也没法优雅退出。非阻塞读配合轮询CPU空转浪费资源。多路复用select/poll/epoll是实际项目里最常用的既能等数据又能设超时还能同时监听多个文件描述符。我个人的选择是单串口采集用poll就够了代码比select清爽没有FD_SETSIZE的限制。如果要同时采集多个串口epoll在大量描述符时性能更好但串口数量一般不会太多poll完全够用。#include poll.h int read_with_timeout(int fd, unsigned char *buf, int maxlen, int timeout_ms) { struct pollfd pfd; pfd.fd fd; pfd.events POLLIN; int ret poll(pfd, 1, timeout_ms); if (ret 0) { perror(poll); return -1; } if (ret 0) { return 0; // 超时无数据 } if (pfd.revents POLLIN) { return read(fd, buf, maxlen); } return -1; }3.2 数据断帧为什么你收到的数据总是缺胳膊少腿串口是字节流没有消息边界的概念。对方设备发一帧20字节的数据你的read可能第一次返回8字节第二次返回12字节也可能两次数据粘在一起返回30字节。这不是bug是串口通信的本质。必须自己实现帧同步逻辑。常见的帧格式有两种定长帧和变长帧。定长帧好办攒够N字节就处理。变长帧通常有帧头、长度字段、数据区、校验字段比如0xAA 0x55 | len | data... | crc。解析的时候需要一个状态机逐字节扫描找到帧头后根据长度字段确定整帧长度凑齐后校验校验通过才交给业务层。typedef enum { STATE_IDLE, STATE_HEADER1, STATE_HEADER2, STATE_LENGTH, STATE_PAYLOAD, STATE_CHECKSUM } parse_state_t; typedef struct { parse_state_t state; unsigned char buf[512]; int index; int expected_len; } frame_parser_t; void parser_init(frame_parser_t *p) { p-state STATE_IDLE; p-index 0; p-expected_len 0; } // 返回1表示解析出一帧完整数据0表示还在解析中 int parser_feed(frame_parser_t *p, unsigned char byte) { switch (p-state) { case STATE_IDLE: if (byte 0xAA) { p-buf[0] byte; p-index 1; p-state STATE_HEADER2; } break; case STATE_HEADER2: if (byte 0x55) { p-buf[1] byte; p-index 2; p-state STATE_LENGTH; } else { p-state STATE_IDLE; } break; case STATE_LENGTH: p-buf[2] byte; p-expected_len byte 4; // 帧头2 长度1 数据len 校验1 p-index 3; p-state STATE_PAYLOAD; break; case STATE_PAYLOAD: p-buf[p-index] byte; if (p-index p-expected_len - 1) { p-state STATE_CHECKSUM; } break; case STATE_CHECKSUM: p-buf[p-index] byte; // 这里做校验假设是简单的累加和 { unsigned char sum 0; for (int i 0; i p-index - 1; i) { sum p-buf[i]; } if (sum byte) { p-state STATE_IDLE; return 1; // 完整帧 } } p-state STATE_IDLE; break; } return 0; }这个状态机的思路是逐字节驱动每收到一个字节就推进状态。好处是不管数据怎么分片到达解析结果都一样。实际项目中我会把read到的数据先塞进一个环形缓冲区然后解析器从环形缓冲区里取字节这样读写解耦逻辑更清晰。3.3 环形缓冲区串口采集的标配组件为什么需要环形缓冲区因为read返回的数据可能包含多帧也可能只有半帧。如果直接在read的缓冲区上做解析处理起来很别扭。环形缓冲区ring buffer提供先进先出的字节流抽象写入端只管往里塞读取端只管从里取两边互不干扰。#define RING_SIZE 4096 typedef struct { unsigned char data[RING_SIZE]; volatile int head; // 写入位置 volatile int tail; // 读取位置 } ring_buf_t; int ring_write(ring_buf_t *r, const unsigned char *src, int len) { int available (r-tail - r-head - 1 RING_SIZE) % RING_SIZE; if (len available) len available; for (int i 0; i len; i) { r-data[r-head] src[i]; r-head (r-head 1) % RING_SIZE; } return len; } int ring_read(ring_buf_t *r, unsigned char *dst, int len) { int used (r-head - r-tail RING_SIZE) % RING_SIZE; if (len used) len used; for (int i 0; i len; i) { dst[i] r-data[r-tail]; r-tail (r-tail 1) % RING_SIZE; } return len; }如果采集线程和解析线程是两个线程head和tail的读写需要考虑内存屏障或者用原子操作。单线程场景下用volatile就够了。这个环形缓冲区的大小要根据数据速率来定115200波特率下每秒最多约11KB数据4KB缓冲区大概能缓冲0.36秒一般够用。如果解析逻辑比较重可以适当加大。4. 数据处理从原始字节到可用信息4.1 字节序和数据类型转换串口传过来的多字节数据字节序大端/小端取决于发送方设备。很多传感器用大端序网络字节序而x86 Linux是小端直接memcpy到int会得到错误的值。必须手动组装// 大端序4字节转uint32 uint32_t be32_to_u32(const unsigned char *p) { return ((uint32_t)p[0] 24) | ((uint32_t)p[1] 16) | ((uint32_t)p[2] 8) | ((uint32_t)p[3]); } // 小端序4字节转uint32 uint32_t le32_to_u32(const unsigned char *p) { return ((uint32_t)p[3] 24) | ((uint32_t)p[2] 16) | ((uint32_t)p[1] 8) | ((uint32_t)p[0]); }浮点数的处理更麻烦有些设备用IEEE 754格式直接传有些用定点数。如果是IEEE 754先按字节序组装成uint32_t再用memcpy转成float。如果是定点数需要知道小数位数比如温度值放大100倍传输收到之后除以100.0。4.2 物理量换算与校准原始ADC值往往需要换算成物理量。比如一个12位ADC参考电压3.3V那么电压值 raw / 4095.0 * 3.3。如果是电流环4-20mA的传感器还要做线性映射。这些换算公式通常写在设备手册里但实际使用中往往需要现场校准——因为传感器有零漂和增益误差。我一般会在程序里留一组校准参数支持通过配置文件或者命令行参数调整。比如温度传感器实测发现比标准温度计低0.5度那就加一个偏移量。这种校准值不要硬编码在代码里否则换个传感器就得重新编译。4.3 数据滤波什么时候该滤什么时候不该滤采集到的数据往往有噪声常见的滤波方法有滑动平均、中值滤波、一阶低通滤波。滑动平均最简单维护一个固定长度的队列每次新数据进来替换最老的输出平均值。中值滤波对脉冲噪声效果好但需要排序计算量稍大。一阶低通滤波适合变化缓慢的信号公式是y alpha * x (1 - alpha) * yalpha越小越平滑但响应越慢。注意滤波会引入延迟对于需要快速响应的控制场景滤波参数要谨慎选择。另外如果原始数据本身就是要保存的建议先保存原始值再保存滤波值这样后期分析时还有原始数据可用。4.4 时间戳别等到保存的时候才想起来每条数据都应该带时间戳而且时间戳应该在数据到达时就打上而不是在保存时。因为从读取到保存之间可能有排队延迟如果保存时才打时间戳时间就不准了。用clock_gettime(CLOCK_MONOTONIC)获取单调递增时间避免系统时间被NTP调整导致时间戳跳变。如果需要和真实时间对应可以在程序启动时记录一次CLOCK_REALTIME和CLOCK_MONOTONIC的差值之后用单调时间推算真实时间。#include time.h struct timespec ts; clock_gettime(CLOCK_MONOTONIC, ts); long long timestamp_ms (long long)ts.tv_sec * 1000 ts.tv_nsec / 1000000;5. 数据保存写文件也有讲究5.1 文本格式还是二进制格式保存格式的选择取决于后续怎么用。如果数据要给人看或者导入Excel分析CSV文本格式最方便。如果数据量大、写入频繁二进制格式更省空间、写入更快。我一般会同时保存两份一份二进制原始数据用于归档和回放一份CSV摘要用于快速查看。CSV格式的坑在于字段里可能有逗号或换行需要做转义。另外浮点数格式化时要注意精度%.6f通常够用但如果是高精度传感器可能需要更多位数。每行末尾用\n还是\r\n取决于目标平台Linux下用\n就行。二进制格式建议自己定义一个简单的文件头记录魔数、版本号、数据记录的结构体大小、采集开始时间等信息方便以后解析。每条记录定长这样可以用fseek快速定位到第N条记录。5.2 缓冲策略fflush的时机很关键标准IO库fopen/fwrite默认是全缓冲的数据先写到用户空间缓冲区缓冲区满了才真正写到磁盘。如果程序崩溃或者断电缓冲区里的数据就丢了。对于采集程序这是不能接受的。解决方案是定期fflush比如每写100条记录或者每隔1秒调用一次fflush。更严格的做法是用O_SYNC标志打开文件每次write都同步到磁盘但这样性能会差很多。折中方案是用fsync定期同步比如每10秒一次。具体间隔取决于你能容忍多少数据丢失。FILE *fp fopen(data.csv, a); // ... 写入数据 ... if (write_count % 100 0) { fflush(fp); }5.3 文件轮转别让单个文件无限增长长时间运行的采集程序如果一直往一个文件里写文件会越来越大最后可能撑爆磁盘。必须做文件轮转。策略有两种按大小轮转和按时间轮转。按大小就是文件超过比如100MB就关闭当前文件新建一个带序号的文件。按时间就是每小时或每天一个文件文件名里带日期时间。char filename[256]; time_t now time(NULL); struct tm *tm_info localtime(now); strftime(filename, sizeof(filename), data_%Y%m%d_%H%M%S.csv, tm_info);轮转的时候要注意关闭旧文件前一定要fflush确保数据落盘。另外如果磁盘空间有限还需要实现旧文件清理逻辑比如只保留最近7天的数据更早的自动删除。删除前最好确认一下文件已经成功写入避免误删。5.4 断电保护采集程序必须考虑的现实问题工控现场断电是家常便饭。如果程序正在写文件时断电文件可能损坏尤其是CSV这种文本格式最后一行可能是半截的。二进制格式相对好一点但最后一条记录也可能不完整。我的做法是每条记录写入前先写一个长度字段或者记录头解析时如果发现记录头不完整或者长度不对就丢弃最后这条不完整记录。另外文件打开用追加模式a这样即使上次异常退出下次启动还能接着写不会覆盖已有数据。如果对数据完整性要求极高可以考虑先写临时文件定期rename成正式文件rename是原子操作能保证文件要么是完整的旧版本要么是完整的新版本。6. 完整代码框架与编译运行6.1 主循环结构把上面的模块串起来主循环大概是这样#include stdio.h #include stdlib.h #include string.h #include signal.h #include poll.h #include time.h #include termios.h #include fcntl.h #include unistd.h static volatile int running 1; void signal_handler(int sig) { (void)sig; running 0; } int main(int argc, char *argv[]) { const char *dev argc 1 ? argv[1] : /dev/ttyUSB0; int fd open_serial(dev, B115200); if (fd 0) return 1; signal(SIGINT, signal_handler); signal(SIGTERM, signal_handler); ring_buf_t ring; memset(ring, 0, sizeof(ring)); frame_parser_t parser; parser_init(parser); FILE *fp fopen(sensor_data.csv, a); if (!fp) { perror(fopen); close(fd); return 1; } unsigned char read_buf[256]; unsigned char byte; int write_count 0; while (running) { int n read_with_timeout(fd, read_buf, sizeof(read_buf), 100); if (n 0) { ring_write(ring, read_buf, n); } // 从环形缓冲区取字节喂给解析器 while (ring_read(ring, byte, 1) 1) { if (parser_feed(parser, byte)) { // 解析出一帧完整数据 // 假设数据区从buf[3]开始长度为expected_len-4 int data_len parser.expected_len - 4; // 这里做业务处理比如解析出温度值 // 然后保存 struct timespec ts; clock_gettime(CLOCK_MONOTONIC, ts); long long ts_ms (long long)ts.tv_sec * 1000 ts.tv_nsec / 1000000; fprintf(fp, %lld,, ts_ms); for (int i 0; i data_len; i) { fprintf(fp, %02X, parser.buf[3 i]); } fprintf(fp, \n); if (write_count % 100 0) { fflush(fp); } } } } fflush(fp); fclose(fp); close(fd); return 0; }编译命令gcc -o serial_collector serial_collector.c -Wall -O2运行./serial_collector /dev/ttyUSB06.2 Python方案什么时候用什么时候不用如果采集频率不高比如每秒几条数据或者只是做原型验证Python的pyserial库非常方便import serial import time import csv ser serial.Serial(/dev/ttyUSB0, 115200, timeout1) with open(data.csv, a, newline) as f: writer csv.writer(f) while True: data ser.read(ser.in_waiting or 1) if data: ts time.monotonic() writer.writerow([ts, data.hex()]) f.flush()Python的缺点是GIL导致多线程性能受限高频采集时可能丢数据。另外Python进程的内存占用和启动时间也比C大。所以我的建议是原型用Python生产用C。如果团队全是Python背景用Python也可以但要注意用serial.read配合合适的超时避免CPU空转。6.3 调试技巧没有硬件怎么测开发阶段不一定手头有真实设备可以用虚拟串口对来模拟。Linux下用socat可以创建一对互相连接的伪终端socat -d -d pty,raw,echo0 pty,raw,echo0它会输出两个设备路径比如/dev/pts/3和/dev/pts/4。你的程序打开其中一个用另一个发数据就能模拟串口通信。发数据可以用echo -e或者写个小脚本。另一个技巧是用strace跟踪系统调用看看read到底返回了什么open是否成功ioctl有没有报错。strace -e traceread,write,open,ioctl ./serial_collector能快速定位问题。7. 那些只有踩过才知道的坑7.1 USB转串口的热插拔问题USB转串口模块拔掉再插上/dev/ttyUSB0可能变成/dev/ttyUSB1原来的文件描述符会失效read返回-1并且errno是ENODEV。程序必须处理这种情况检测到ENODEV或者EIO时关闭旧fd重新扫描设备尝试重新打开。如果设备名是固定的通过udev规则重新打开同一个路径就行。如果不是需要重新扫描/dev/serial/by-id/。重连逻辑要加退避不能一失败就疯狂重试否则CPU和日志都会爆炸。我一般用指数退避第一次等1秒第二次2秒最多等30秒。7.2 数据量太大导致磁盘写满这个坑很隐蔽程序跑得好好的突然某天发现不写数据了一查磁盘满了。除了文件轮转和旧文件清理还应该监控磁盘剩余空间低于阈值时告警或者暂停采集。可以用statvfs获取磁盘信息#include sys/statvfs.h struct statvfs vfs; statvfs(/data, vfs); unsigned long free_mb vfs.f_bavail * vfs.f_frsize / (1024 * 1024); if (free_mb 100) { // 告警或停止写入 }7.3 串口参数被意外修改有些设备在上电或者复位时会发送一些控制字符如果串口配置了ICRNL把\r转成\n或者IXON软件流控这些字符可能被解释成控制命令导致串口行为异常。所以配置串口时一定要把所有不必要的输入处理关掉用原始模式raw mode。我在代码里把c_iflag和c_lflag里能关的都关了就是为了避免这种意外。7.4 多线程下的资源竞争如果采集和保存分在两个线程共享的环形缓冲区需要加锁或者用无锁队列。加锁简单但可能影响性能无锁队列性能好但实现复杂容易出错。我的经验是如果采集频率在每秒几千条以内用互斥锁完全够用不要过早优化。真正需要无锁队列的场景通常是每秒几十万条以上的高速采集那种场景一般也不会用串口。7.5 日志和数据的分离程序运行日志比如打开串口成功、解析错误和数据文件要分开保存。日志用syslog或者单独的日志文件数据用专门的数据文件。不要把日志和数据混在一个文件里否则后期解析数据时还要过滤日志行很麻烦。日志级别也要控制正常运行时只记warning以上调试时才开debug否则日志文件增长太快。8. 性能调优的几个实际手段8.1 减少系统调用次数每次read都是一次系统调用有开销。如果数据速率很高可以一次read尽量多的数据。VMIN和VTIME的设置会影响read的行为配合poll使用可以在有数据时批量读取。另外read的缓冲区不要太小256字节到4KB之间比较合适。8.2 文件IO的批量写入前面提到用fprintf逐条写每次fprintf都有格式化开销。如果性能要求高可以自己拼字符串到缓冲区攒够一批再write。或者用writev一次性写入多个缓冲区。对于二进制格式直接fwrite结构体数组比逐条写快得多。8.3 CPU占用率优化如果发现程序CPU占用率高先检查是不是在忙等待。poll的超时时间不要设0设0会立即返回变成轮询。设100ms到1s比较合理。另外解析器逐字节处理在大数据量时可能成为瓶颈可以优化成批量扫描帧头找到帧头后再逐字节解析。8.4 内存占用控制环形缓冲区、解析缓冲区、文件写缓冲区的大小都要合理设置。嵌入式设备内存有限不要动不动就分配几MB的缓冲区。根据实际数据速率计算所需的最小缓冲区大小留2-3倍余量即可。9. 从采集到可用数据的完整链路回顾整条链路走下来核心就三件事稳定地读、正确地解析、可靠地存。读的时候注意设备识别、权限、参数配置、超时控制解析的时候注意帧同步、字节序、校验、时间戳存的时候注意格式选择、缓冲策略、文件轮转、断电保护。每一个环节都有坑但每一个坑都有成熟的解决方案。我在实际项目里最深的体会是串口采集程序的健壮性80%取决于异常处理做得好不好。正常流程谁都能写但设备拔插、磁盘满、数据异常、程序被杀这些情况处理不好程序就没法长期稳定运行。所以写这类程序一定要把出错了怎么办想清楚而不是只想着正常情况怎么跑。另外测试阶段一定要用真实设备跑足够长的时间至少24小时以上观察内存有没有泄漏、文件有没有正常轮转、时间戳有没有跳变。很多问题只有在长时间运行后才会暴露出来。如果条件允许模拟断电、拔插串口、填满磁盘这些异常场景验证程序的恢复能力。这些测试做完程序才算真正可用。