1. 项目缘起与整体设计思路
1.1 为什么要在ESP32上折腾ModbusTCP分片缓存
先说结论:如果你用ESP32做ModbusTCP从站,且寄存器数量超过200个,不做分片缓存,迟早会遇到响应超时、数据错乱、客户端反复重连的问题。这不是危言耸听,是我在实际项目中踩了三次坑之后才彻底搞明白的事。
ESP32作为一款低成本、带Wi-Fi的MCU,在工业数据采集场景里出场率极高。很多中小型项目用它做ModbusTCP网关,把下面的RS485设备数据汇聚上来,再通过Wi-Fi转发给上位机或SCADA系统。问题就出在这个“汇聚”环节——ModbusTCP的单个请求报文有长度限制,TCP层虽然可以分片,但Modbus协议本身对PDU(协议数据单元)有明确的大小约束。当你需要一次性读取几百个保持寄存器时,一个请求塞不下,客户端要么分批请求,要么就得靠服务端做分片缓存来平滑处理。
我最初的做法很粗暴:客户端每次请求多少,我就实时去采集多少。结果就是响应时间忽长忽短,遇到慢速从站直接超时。后来改成“后台轮询+缓存映射”的架构,才把稳定性拉上来。这个项目的核心,就是围绕ESP32的内存特性和ModbusTCP的协议约束,设计一套高效、可靠的分片缓存机制。
1.2 整体架构:三层缓存模型
我把整个系统拆成三层,从下往上依次是采集层、缓存层、服务层。采集层负责通过UART或SPI跟下位设备通信,把原始数据读上来;缓存层是核心,用分片的方式把寄存器数据映射到ESP32的RAM里;服务层跑ModbusTCP服务器,响应来自Wi-Fi的请求。
为什么是三层而不是两层?因为ESP32的RAM太宝贵了。ESP32-WROOM-32典型可用堆内存也就200KB出头,如果开Wi-Fi和TCP协议栈,实际能拿来存业务数据的可能不到100KB。你要缓存1000个保持寄存器,每个寄存器2字节,那就是2KB,看起来不多,但加上分片管理的元数据、双缓冲、队列,内存开销会翻好几倍。分层之后,采集层和缓存层可以独立优化,服务层只管从缓存里取数,不用关心数据是怎么来的。
分片的核心思想是:不把寄存器当成一整块连续内存来管理,而是切成固定大小的“片”,每片比如32个寄存器。每个片有独立的版本号、时间戳、锁标记。客户端请求某个地址范围时,服务层先算出涉及哪些片,然后逐片检查有效性,无效的片触发采集层刷新,有效的片直接返回。这样既避免了全量刷新的开销,又保证了数据的新鲜度。
1.3 方案选型背后的取舍
有人可能会问:为什么不直接用FreeRTOS的队列或者信号量来同步?为什么要自己搞一套分片缓存?
我试过用队列。每个寄存器一个队列项,采集任务往里塞,服务任务往外取。问题是队列深度有限,寄存器一多,队列就爆了。而且队列是FIFO的,服务端要读地址0x1000的数据,结果队列里全是0x2000之后的,你得把前面的都出队再丢弃,效率极低。
也试过用全局数组加互斥锁。简单是简单,但锁的粒度太大。采集任务写数组的时候,服务任务只能干等。如果采集周期是100ms,服务端请求频率是10Hz,那每次请求都可能撞上锁,延迟直接翻倍。
分片缓存的好处在于锁的粒度细。每个片独立加锁,采集任务写片A的时候,服务任务可以读片B。而且片的大小可以调,32个寄存器一片,锁冲突概率就降到了1/32以下。实测下来,同样硬件条件下,分片方案的请求响应时间从平均45ms降到了12ms,最坏情况也从200ms降到了35ms。
另一个取舍是分片大小。片太小,元数据开销大,1000个寄存器切成32片,每片的管理结构至少20字节,总共640字节,还能接受。片太大,比如128个寄存器一片,锁冲突概率上去了,而且一次刷新要等更久。我最后选了32,是在内存开销和并发性能之间折中的结果。你可以根据自己项目里寄存器的访问局部性来调——如果客户端总是连续读一大片,那就把片调大;如果访问很分散,那就调小。
2. 核心细节解析与实操要点
2.1 分片数据结构的设计
先看核心结构体。我用C语言写的,跑在ESP-IDF环境下,但思路是通用的,你用Arduino框架也能照搬。
typedef struct { uint16_t start_addr; // 片起始寄存器地址 uint16_t count; // 片内寄存器数量 uint16_t *data; // 数据缓冲区指针 uint32_t version; // 版本号,每次刷新递增 uint64_t last_update; // 上次刷新时间戳(微秒) SemaphoreHandle_t lock; // 片级互斥锁 bool valid; // 数据是否有效 } modbus_slice_t;这里有几个关键点。version字段是给服务层用的,客户端如果带了“最小版本号”要求,服务层可以快速判断缓存是否满足。last_update用来做超时淘汰,比如超过5秒没刷新的片,标记为无效,下次请求时强制刷新。lock是FreeRTOS的二值信号量,不是互斥锁,因为我不需要优先级继承,只需要简单的互斥。
数据缓冲区data是动态分配的,不是静态数组。为什么?因为ESP32的DRAM分两块,一块是内部的,一块是外部的(如果接了PSRAM)。静态数组会占用宝贵的内部RAM,动态分配可以灵活选择。如果你的片数据访问频繁,放内部RAM;如果只是偶尔读,放PSRAM也行。我一般把前16个片放内部,后面的放外部,因为客户端通常更关注低地址区域。
2.2 地址映射与片索引计算
客户端发来一个请求,起始地址0x1000,数量50。怎么快速定位到涉及的片?
#define SLICE_SIZE 32 int get_slice_index(uint16_t addr) { return addr / SLICE_SIZE; } void get_slice_range(uint16_t start, uint16_t count, int *first, int *last) { *first = start / SLICE_SIZE; *last = (start + count - 1) / SLICE_SIZE; }看起来简单,但有个坑:寄存器地址不一定从0开始。Modbus协议里,保持寄存器通常从40001开始编,但实际传输时地址是0-based的。你得先统一到0-based,再做除法。我见过有人直接拿40001去除32,结果算出来的索引巨大,数组直接越界。
另一个坑是跨片请求。起始地址0x1000,数量50,SLICE_SIZE是32。第一个片是0x1000到0x101F,第二个片是0x1020到0x103F。请求覆盖了第一个片的全部和第二个片的前18个。服务层需要把请求拆成两段,分别从两个片里取数据,再拼成一个响应。拼接的时候注意字节序,Modbus是大端,ESP32是小端,每个寄存器都要做htons转换。
注意:跨片请求的拼接缓冲区要提前分配好,不要在中断里malloc。我一般用一个静态的512字节缓冲区,够放256个寄存器。如果请求超过256个寄存器,直接返回异常码,让客户端分批请求。
2.3 双缓冲与无锁读取
服务层读数据的时候,如果采集层正在写同一个片,怎么办?加锁当然可以,但锁会阻塞。我的做法是双缓冲:每个片有两个数据缓冲区,采集层写A的时候,服务层读B;写完了交换指针。
typedef struct { uint16_t *buf_a; uint16_t *buf_b; uint16_t *active; // 当前采集层写入的缓冲区 uint16_t *ready; // 当前服务层可读的缓冲区 } double_buffer_t;交换指针的操作必须是原子的。在ESP32上,可以用__atomic_exchange_n或者关中断。我一般用临界区,因为交换频率不高,每次刷新才交换一次,开销可以忽略。
双缓冲的代价是内存翻倍。1000个寄存器,原来2KB,现在4KB。但换来的是服务层读取完全无锁,响应时间更稳定。如果你的内存实在紧张,可以只对高频访问的片用双缓冲,低频片还是用锁。
2.4 采集任务的调度策略
采集层怎么知道该刷新哪个片?两种模式:轮询模式和按需模式。
轮询模式是后台任务定期遍历所有片,不管有没有人读,都去采集。优点是数据总是新鲜的,服务层响应快。缺点是浪费带宽和CPU,尤其是那些没人读的片。
按需模式是服务层发现片无效时,发个信号给采集任务,采集任务再去刷。优点是省资源,缺点是第一次请求会有延迟。
我最后用的是混合模式:高频片(比如前8个)轮询,低频片按需。怎么判断高频?统计每个片的请求次数,超过阈值的加入轮询列表。这个统计可以放在服务层做,每次响应完更新计数器。
采集任务的优先级要低于Wi-Fi任务,但高于空闲任务。我设的是优先级5(Wi-Fi通常是6或7)。栈大小给4096字节,因为要调用UART驱动和可能的内存分配。
3. 实操过程与核心环节实现
3.1 环境搭建与依赖配置
硬件清单:ESP32-WROOM-32开发板一块,USB转TTL一个,如果接RS485设备还需要一个MAX485模块。软件用ESP-IDF v4.4以上,或者Arduino框架加ModbusIP库。我下面以ESP-IDF为例,因为对内存和任务的控制更精细。
首先创建项目:
idf.py create-project modbus_tcp_slice cd modbus_tcp_slice idf.py menuconfig在menuconfig里要改几个地方。Component config → FreeRTOS → Tick rate 改成1000Hz,这样时间戳精度到毫秒。Component config → ESP32-specific → Support for external, SPI-connected RAM 如果用了PSRAM就打开。Component config → LWIP → TCP → TCP receive window 设成最大,因为ModbusTCP响应可能比较大。
然后配置Wi-Fi。我一般把SSID和密码放在Kconfig里,不硬编码在源码中。在项目根目录建一个Kconfig.projbuild:
menu "ModbusTCP Slice Config" config WIFI_SSID string "WiFi SSID" default "your_ssid" config WIFI_PASSWORD string "WiFi Password" default "your_password" config SLICE_SIZE int "Registers per slice" default 32 config MAX_SLICES int "Maximum number of slices" default 64 endmenu这样编译时可以通过menuconfig改,也可以直接在sdkconfig里改。
3.2 分片缓存的初始化
初始化分两步:先分配片数组,再分配每个片的数据缓冲区。
modbus_slice_t *slices; int slice_count; void slice_cache_init(uint16_t total_regs) { slice_count = (total_regs + SLICE_SIZE - 1) / SLICE_SIZE; slices = calloc(slice_count, sizeof(modbus_slice_t)); for (int i = 0; i < slice_count; i++) { slices[i].start_addr = i * SLICE_SIZE; slices[i].count = SLICE_SIZE; slices[i].data = heap_caps_malloc(SLICE_SIZE * 2, MALLOC_CAP_8BIT); slices[i].version = 0; slices[i].last_update = 0; slices[i].valid = false; slices[i].lock = xSemaphoreCreateBinary(); xSemaphoreGive(slices[i].lock); } }注意最后一个片的count可能小于SLICE_SIZE,要单独处理。heap_caps_malloc的第二个参数指定内存类型,MALLOC_CAP_8BIT表示任意8位可访问内存,如果要用PSRAM就加MALLOC_CAP_SPIRAM。
提示:
calloc和heap_caps_malloc的返回值一定要检查。ESP32内存紧张,分配失败是常态。我一般会在分配失败时打印剩余堆大小,方便排查。
3.3 ModbusTCP服务端的实现
服务端基于lwIP的socket API,监听502端口。每个客户端连接创建一个任务,任务里循环接收请求、解析、从缓存取数、组包、发送。
void modbus_server_task(void *arg) { int sock = *(int *)arg; uint8_t req[256], resp[512]; while (1) { int len = recv(sock, req, sizeof(req), 0); if (len <= 0) break; // 解析MBAP头 uint16_t trans_id = (req[0] << 8) | req[1]; uint16_t proto_id = (req[2] << 8) | req[3]; uint16_t pdu_len = (req[4] << 8) | req[5]; uint8_t unit_id = req[6]; uint8_t func_code = req[7]; if (func_code == 0x03) { // 读保持寄存器 uint16_t start = (req[8] << 8) | req[9]; uint16_t count = (req[10] << 8) | req[11]; // 从分片缓存取数 int first, last; get_slice_range(start, count, &first, &last); uint16_t *out = (uint16_t *)(resp + 9); int out_idx = 0; for (int i = first; i <= last; i++) { xSemaphoreTake(slices[i].lock, portMAX_DELAY); if (!slices[i].valid) { // 触发刷新,这里简化处理,实际要发信号给采集任务 refresh_slice(i); } // 计算片内偏移 int offset = (start > slices[i].start_addr) ? (start - slices[i].start_addr) : 0; int n = slices[i].count - offset; if (n > count - out_idx) n = count - out_idx; for (int j = 0; j < n; j++) { out[out_idx++] = htons(slices[i].data[offset + j]); } xSemaphoreGive(slices[i].lock); } // 组响应 resp[0] = req[0]; resp[1] = req[1]; resp[2] = 0; resp[3] = 0; resp[4] = 0; resp[5] = 3 + count * 2; resp[6] = unit_id; resp[7] = func_code; resp[8] = count * 2; send(sock, resp, 9 + count * 2, 0); } } close(sock); vTaskDelete(NULL); }这段代码有几个关键点。第一,recv的缓冲区256字节够放大多数请求,但如果你要支持写多个寄存器,得加大。第二,htons转换不能漏,否则客户端读到的数据是字节反的。第三,刷新片的操作不能阻塞太久,我这里是简化写法,实际应该发消息给采集任务,让采集任务异步刷新。
3.4 采集任务的实现
采集任务负责跟下位设备通信,把数据填到片里。我用的是UART,波特率9600,8N1。如果是RS485,还要控制DE/RE引脚。
void采集_task(void *arg) { uint8_t uart_buf[256]; while (1) { // 等待刷新信号或超时 ulTaskNotifyTake(pdTRUE, pdMS_TO_TICKS(100)); for (int i = 0; i < slice_count; i++) { if (slices[i].valid && (esp_timer_get_time() - slices[i].last_update) < 5000000) { continue; // 5秒内刷过的跳过 } // 构造Modbus RTU请求 uint8_t cmd[8]; cmd[0] = 0x01; // 从站地址 cmd[1] = 0x03; cmd[2] = slices[i].start_addr >> 8; cmd[3] = slices[i].start_addr & 0xFF; cmd[4] = 0; cmd[5] = slices[i].count; uint16_t crc = modbus_crc(cmd, 6); cmd[6] = crc & 0xFF; cmd[7] = crc >> 8; uart_write_bytes(UART_NUM_1, cmd, 8); // 读响应 int len = uart_read_bytes(UART_NUM_1, uart_buf, sizeof(uart_buf), pdMS_TO_TICKS(200)); if (len >= 5 + slices[i].count * 2) { xSemaphoreTake(slices[i].lock, portMAX_DELAY); for (int j = 0; j < slices[i].count; j++) { slices[i].data[j] = (uart_buf[3 + j*2] << 8) | uart_buf[4 + j*2]; } slices[i].version++; slices[i].last_update = esp_timer_get_time(); slices[i].valid = true; xSemaphoreGive(slices[i].lock); } } } }这里有个细节:ulTaskNotifyTake的超时设成100ms,意味着即使没有刷新信号,每100ms也会轮询一次。这是为了处理那些按需刷新的片。如果你全用轮询模式,可以把超时设长一点,比如1秒。
CRC校验函数是标准Modbus CRC16,网上有很多实现,我就不贴了。注意CRC的低字节在前,高字节在后,跟数据字节序相反。
3.5 内存优化与PSRAM使用
ESP32的内部RAM分DRAM和IRAM,DRAM大概160KB可用。如果你要缓存2000个寄存器,双缓冲就是8KB,加上片元数据,大概10KB。看起来不多,但Wi-Fi协议栈要占几十KB,TCP窗口又要占几十KB,实际剩给业务的可能就50KB左右。
如果不够用,可以启用PSRAM。ESP32-WROVER模组带4MB PSRAM,通过SPI接口访问,速度比内部RAM慢,但容量大。把片数据放PSRAM,元数据放内部RAM,是个不错的折中。
slices[i].data = heap_caps_malloc(SLICE_SIZE * 2, MALLOC_CAP_SPIRAM | MALLOC_CAP_8BIT);注意PSRAM不能用于DMA,所以UART的接收缓冲区不能用PSRAM。另外,PSRAM的访问延迟大概比内部RAM高3到5倍,如果服务层读取非常频繁,可能会成为瓶颈。我的做法是给每个片加一个“热度”计数器,热度高的片迁移到内部RAM,热度低的迁到PSRAM。迁移操作在采集任务的空闲时间做,不影响实时性。
注意:启用PSRAM后,
heap_caps_get_free_size(MALLOC_CAP_SPIRAM)可以查看剩余PSRAM。如果返回0,说明PSRAM没初始化成功,检查menuconfig里的PSRAM选项和硬件连接。
4. 常见问题与排查技巧实录
4.1 响应超时与数据错乱
现象:客户端读50个寄存器,有时候正常,有时候超时,有时候数据对不上。
排查思路:先看是不是跨片了。50个寄存器,SLICE_SIZE=32,肯定跨两个片。如果两个片的刷新时间差很大,比如第一个片是刚刷的,第二个片是5秒前刷的,那拼出来的数据就是“半新半旧”。客户端如果做数据校验,就会报错。
解决:给每个片加一个“批次号”,服务层拼数据时检查所有涉及的片批次号是否一致。不一致就触发全量刷新,等所有片都刷新完再返回。或者更简单:把跨片请求拆成多个单片的Modbus响应,让客户端自己拼。但这样客户端要改,不太现实。
我最后的做法是:服务层发现跨片且批次不一致时,发信号给采集任务,采集任务优先刷新这些片,服务层等待一个信号量,超时200ms。实测下来,99%的请求能在50ms内完成。
速查表:
| 现象 | 可能原因 | 解决方法 |
|---|---|---|
| 响应超时 | 片锁被采集任务长时间占用 | 减小片大小,降低锁持有时间 |
| 数据错乱 | 跨片批次不一致 | 加批次号校验,强制同步刷新 |
| 客户端反复重连 | 响应长度超过TCP窗口 | 增大TCP窗口,或让客户端分批请求 |
| 内存分配失败 | 片数据太多,堆耗尽 | 启用PSRAM,或减少缓存寄存器数量 |
4.2 采集任务饿死
现象:服务层请求频繁时,采集任务一直拿不到CPU,数据越来越旧。
原因:服务层任务优先级设得太高,或者采集任务在等UART时被阻塞太久。
解决:把采集任务优先级调到比服务层高一级。服务层任务在等待片锁时,用xSemaphoreTake带超时,超时后主动让出CPU。另外,UART读取的超时不要设太长,200ms足够,设1秒的话采集任务会卡死。
我还在采集任务里加了个看门狗:每次循环开始喂狗,如果超过2秒没喂,系统重启。这样即使采集任务卡死,也能自动恢复。
4.3 Wi-Fi干扰导致的丢包
现象:有线测试一切正常,一开Wi-Fi就偶尔丢包,ModbusTCP响应时有时无。
原因:ESP32的Wi-Fi和UART共用DMA通道,Wi-Fi流量大时,UART接收可能丢字节。
解决:把UART的接收缓冲区加大到512字节,并且用中断方式接收,不要在任务里轮询。另外,Wi-Fi的省电模式关掉,esp_wifi_set_ps(WIFI_PS_NONE),虽然功耗高一点,但稳定性好很多。
如果还是丢包,可以在应用层加重试。ModbusTCP本身没有重试机制,但客户端通常会重发。服务层要能处理重复的请求,不要因为收到重复请求就乱套。我的做法是给每个请求算个哈希,缓存最近10个请求的响应,重复请求直接返回缓存。
4.4 片大小选择的经验法则
片大小不是拍脑袋定的。我总结了一个简单的公式:
片大小 = 客户端平均请求寄存器数 / 2比如客户端通常一次读50个,那SLICE_SIZE就设25,取整到32。这样大多数请求只涉及1到2个片,跨片概率低。
如果客户端请求数波动很大,有时候读10个,有时候读200个,那就取中位数。或者做两级分片:大请求用大片,小请求用小片。但这样复杂度上去了,一般项目没必要。
我实测过SLICE_SIZE从8到128的各种情况,32在ESP32上综合表现最好。内存开销可接受,锁冲突概率低,跨片拼接也不复杂。
4.5 调试技巧:用GPIO打点测时间
ESP32没有示波器的时候,可以用GPIO翻转来测时间。在服务层收到请求时拉高一个引脚,发送响应后拉低,用逻辑分析仪或者另一个ESP32来测脉宽。
gpio_set_level(GPIO_NUM_2, 1); // ... 处理请求 ... gpio_set_level(GPIO_NUM_2, 0);我一开始用esp_timer_get_time()打时间戳,但打印日志本身有开销,测不准。GPIO翻转几乎零开销,测出来的时间更真实。后来发现,从收到请求到发出响应,平均12ms,最坏35ms,跟我的预期一致。
这个技巧在优化锁粒度的时候特别有用。你可以分别测加锁前、加锁后、解锁后的时间,看看锁到底占了多少。我测下来,锁本身的开销不到1ms,大部分时间花在UART读取和CRC计算上。
4.6 版本号回绕的处理
version字段是uint32_t,理论上会回绕。回绕周期是2的32次方,按每秒刷新100次算,也要一年多才回绕一次。但工业设备可能连续跑几年,所以还是要处理。
我的做法是:比较版本号时不用>,用(int32_t)(a - b) > 0。这样即使回绕,也能正确判断新旧。这是处理无符号计数器回绕的标准技巧,Linux内核里大量使用。
static inline bool version_newer(uint32_t a, uint32_t b) { return (int32_t)(a - b) > 0; }4.7 客户端兼容性测试
不同厂家的ModbusTCP客户端行为不一样。有的客户端一次只读1个寄存器,有的读125个(协议上限)。有的客户端会发广播地址0,有的只发单播。有的客户端超时设得很短,500ms就断。
我建议至少测三种客户端:一种桌面调试工具,一种PLC自带的客户端,一种自己写的Python脚本。Python脚本用pymodbus库,几行代码就能测:
from pymodbus.client import ModbusTcpClient client = ModbusTcpClient('192.168.1.100', port=502) rr = client.read_holding_registers(0, 50, unit=1) print(rr.registers) client.close()如果pymodbus能稳定读到数据,基本就没什么大问题了。如果读不到,先抓包看请求有没有发出去,再看响应有没有回来。Wireshark过滤tcp.port == 502,一目了然。
4.8 长期运行的稳定性建议
工业现场要求7x24小时运行,ESP32的稳定性需要额外关注。我一般加三个机制:
第一,内存泄漏检测。每次分配和释放都记录,定期打印剩余堆大小。如果发现堆持续下降,肯定有泄漏。esp_get_free_heap_size()可以随时查。
第二,任务栈溢出检测。FreeRTOS有栈溢出钩子,在menuconfig里打开CONFIG_FREERTOS_CHECK_STACKOVERFLOW,设成Canary模式。一旦溢出,系统会打印哪个任务出了问题。
第三,看门狗。除了任务看门狗,我还加了一个硬件看门狗,用ESP32内置的RTC看门狗,超时10秒。如果主循环卡死,10秒后自动重启。
这三个机制加上去,连续跑一个月不重启是常态。我有个项目在客户现场跑了半年多,中间只因为停电重启过一次。
4.9 性能优化的小技巧
最后分享几个实测有效的优化点。第一,CRC计算用查表法,比逐位计算快5倍以上。表可以放在Flash里,不占RAM。第二,UART接收用DMA,不要用中断逐字节收,DMA收完一帧再触发中断,CPU占用率从15%降到3%。第三,服务层的响应缓冲区用静态分配,不要每次malloc,避免内存碎片。
还有一个容易被忽略的点:Wi-Fi的DTIM间隔。默认是3,改成1可以降低延迟,但功耗增加。如果是插电设备,改成1;如果是电池设备,保持3。
这些优化加起来,ESP32在跑ModbusTCP服务端的同时,还能剩50%以上的CPU空闲,足够再跑一个Web服务器或者MQTT客户端。我现在的项目里,ESP32同时跑ModbusTCP、MQTT、OTA,三个任务并行,稳定得很。