风电监控这种项目很多时候外行看着高大上内行一看全是脏活累活。说是物联网其实代码翻来覆去就是跟传感器数据死磕采集、清洗、流转、落地、展示。最近我手上这个Spring Boot风电监控项目正好把这条链路完整跑了一遍风机转速、齿轮箱温度、振动幅值这些高频数据每天几千万条往上走。今天不绕弯子直接把代码里的数据处理思路拆开聊看完你也能拿这套逻辑去套自己的设备接入场景。这个项目能解决的问题很典型设备协议杂、数据频次高、实时性要求严同时还要兼顾后续的数据分析和报警联动。我会从接入层、处理层、推送层、存储层一条线往下讲中间穿插实际踩过的坑和优化手段。对于正在搞Spring Boot物联网项目、尤其是做工业设备数据采集的同学这篇文章应该能帮你省掉不少试错时间。1. 先说清楚这个风电监控项目到底长什么样1.1 设备端和数据特点风电场里每台风机都是一个独立的工业设备集合体。机舱里有主轴转速传感器、齿轮箱温度传感器、振动传感器塔基还有电能质量监测模块。每台风机大概几十个测点一个风场几十台上百台风机加起来监测点几千个。这些传感器传来的数据分两类一类是慢速状态量比如舱内环境温度、油压几秒钟一条就够了另一类是快速过程量比如主轴转速和齿轮箱振动这类数据在风机启停、变桨、偏航时变化剧烈需要毫秒级或者百毫秒级采样。这套系统对后端最直接的压力就是高频数据的吞吐能力和实时链路稳定性。上报频次按照每台风机每秒一次、每次几十个测点的规模估算一个百台机组的风场每秒要处理几千条数据点高峰期还会翻倍。Spring Boot在这个场景下并不吃亏只要不在业务代码里写阻塞操作数据进来直接异步丢队列处理完全扛得住。1.2 Spring Boot做物联网后端的基本架构做工业物联网很多团队喜欢一上来就上重型中间件Kafka、Flink、时序数据库全堆上。我的这套架构比较朴素核心就三块Spring Boot Netty负责设备接入和报文解析自研消息分发组件负责把解析后的数据分发给下行存储和实时推送MySQL分表 Redis缓存先满足业务跑通数据量大了再考虑迁移时序库整套逻辑就是设备通过TCP长连接把采集报文推送到Netty网关Netty解码后把数据对象交给Spring容器管理的Service层Service层统一处理后写入存储同时通过WebSocket实时推送到监控大屏。设备端TCP/Modbus协议 ↓ Netty接入网关报文解码、心跳管理 ↓ Spring Boot Service业务校验、异常值清洗、转速温度关联处理 ↓ 存储层(MySQL分表) 实时推送(WebSocket/Redis)这套链路里最关键的环节不是Netty也不是MySQL而是中间那层Spring Boot Service的数据流转设计。数据从原始报文变成业务可用、存储友好、推送即时的结构体这个转化过程才是“骚操作”集中区。2. 传感器数据上来的第一道关卡接入层的那些破事2.1 Netty Spring Boot怎么配合不打架Netty是独立于Spring Boot的通信框架两者结合最常见的问题就是Bean管理冲突。设备接入的Handler里需要调用Spring管理的业务Service但如果直接在Handler里使用new去创建Service这个对象就是脱离Spring容器管理的野对象事务、AOP切面、连接池统统失效。我的做法是写一个SpringContextHolder在Netty模块启动前用ApplicationContext预先把需要的Bean注册进去Component public class SpringContextHolder implements ApplicationContextAware { private static ApplicationContext applicationContext; Override public void setApplicationContext(ApplicationContext applicationContext) { SpringContextHolder.applicationContext applicationContext; } public static T T getBean(ClassT clazz) { return applicationContext.getBean(clazz); } }然后在Netty的Handler里通过SpringContextHolder.getBean(DeviceDataService.class)来获取业务Bean。这样设备通道里的回调方法就能直接调用Spring的事务方法数据解析和入库的链路完整串起来了。设备接入的Handler里另一个容易踩的坑是不要在channelRead方法里直接做耗时操作。Netty的I/O线程是很宝贵的资源一个Handler卡住了后面所有连接的报文都要排队等。所以我在Handler里只做解码和对象转换然后立刻把数据交给线程池处理Override protected void channelRead0(ChannelHandlerContext ctx, String msg) { // 解码报文为设备数据对象 DeviceRawData rawData protocolDecoder.decode(msg); // 丢给业务线程池异步处理不阻塞IO线程 dataProcessExecutor.execute(() - deviceDataService.handleDeviceData(rawData)); }线程池用Spring的Bean定义核心线程数、队列容量都按设备规模和高峰突发量做了估算。2.2 数据帧解析中的位运算细节风机传感器报文里很多数据不是简单的整数类型。Modbus协议里常见的是两个字节表示一个16位的数值有的协议还讲究高低字节序、补码表示负数。比如风机转速设备端上报的两个字节是0x12 0x34。在这个客户端的协议里是低位在前实际转速值应该是合并后除以10保留一位小数。解析代码长这样public double parseSpeed(byte[] data, int offset) { // 低位在前data[offset]是低字节data[offset1]是高字节 int rawValue (data[offset] 0xFF) | ((data[offset 1] 0xFF) 8); // 如果设备用补码表示负数这里还要判断符号位 if ((rawValue 0x8000) ! 0) { rawValue rawValue - 0x10000; } return rawValue / 10.0; } 0xFF是为了防止Java中byte转int时发生符号扩展这个坑新手基本都会踩。之前有一次温度6023.5度的问题就是因为温度正常情况下应该用无符号字节解析代码里却用了有符号转换负温度被当成巨大正数。位运算解析没有捷径就是对照协议文档一个字节一个字节抠。我整理了一个小表放在项目文档里排查数据解析问题时能少费不少神数据类型字节序符号处理缩放系数风机转速低前高后有符号0.1齿轮箱温度低前高后无符号0.1振动加速度高前低后有符号0.001瞬时功率低前高后无符号1.02.3 设备注册表和在线状态管理接入层还有一个容易被低估的功能就是设备在线管理。风场网络环境不比机房光纤被挖断、设备断电重启都时有发生设备不在线时需要第一时间感知并在大屏上预警。我通过Netty的ChannelGroup管理所有建立的连接再用一个ConcurrentHashMap把设备编号和Channel做一一映射Component public class DeviceChannelRegistry { private final ConcurrentHashMapString, Channel channelMap new ConcurrentHashMap(); private final ConcurrentHashMapString, Instant lastHeartbeatMap new ConcurrentHashMap(); public void register(String deviceId, Channel channel) { channelMap.put(deviceId, channel); lastHeartbeatMap.put(deviceId, Instant.now()); } public boolean isOnline(String deviceId) { Channel channel channelMap.get(deviceId); return channel ! null channel.isActive(); } public void heartbeat(String deviceId) { lastHeartbeatMap.put(deviceId, Instant.now()); } }心跳机制上设备每10秒发一次心跳报文如果我120秒没收到一台设备的心跳就认为它掉线了触发掉线通知。这里有个细节要注意心跳处理一定要和业务数据解析分开因为心跳包频率高、消息格式简单混在一起解析会让代码变得不可维护。3. 每秒几百条数据进来后怎么做到不卡不丢不串3.1 数据平滑别让毛刺数据骗了你的眼睛传感器数据尤其是振动和转速这类物理量原始信号里往往有噪声。设备本身的机械抖动、电磁干扰、ADC采样误差都会产生毛刺数据。如果后端拿到什么存什么、拿什么推什么监控大屏上就会出现频繁跳变的虚假报警。我的处理和风机控制逻辑不一样控制逻辑要求实时响应但监控系统可以容忍几十毫秒的延迟换取数据平滑。这里我用的是一阶滞后滤波本质上就是加权平均public class DataFilter { private final double alpha; private volatile double lastValue; public DataFilter(double alpha) { this.alpha alpha; // 比如0.2表示新数据权重20% } public double apply(double newValue) { lastValue alpha * newValue (1 - alpha) * lastValue; return lastValue; } }alpha的取值比较讲究太快起不到平滑作用太慢数据响应又太迟钝。对风机转速这种变化相对平缓的量我取0.15到0.2对振动这种波动剧烈的量取0.05到0.1。刚开始我对所有测点统一用0.1的alpha结果转速探头上数据响应像肉盾一样迟钝风机都升速到1800转了显示还在1300转趴着。后来才意识到不同物理量对滤波系数的要求完全不一样。3.2 异常值清洗剔除假数据保留真异常滤波解决了数据毛刺但还有一种情况需要单独处理报文解析出来几倍于正常范围的离谱数值。比如温度正常在60到80度突然蹦出个402.6度这显然是传感器故障或通信干扰不是真实故障。清洗规则我用的是变化率限制 历史分位数双重校验。变化率指的是相邻两条数据的变化幅度不能超过物理极限风机转速每秒最多升降200转超过这个范围就判定为异常值。public class AbnormalValueFilter { private double maxChangeRate; public double filter(double previousValue, double currentValue) { if (Math.abs(currentValue - previousValue) maxChangeRate) { // 超限值不直接入库用上一次的有效值填充 return previousValue; } return currentValue; } }分位数校验这边我维护每个测点最近1000条有效数据的分位数分布当前值超过P99.9即99.9%的历史数据都在这个阈值内且持续3条以上才认为是真实异常。这样处理的好处是单个瞬时尖峰不会触发报警但持续攀爬的温度增长一定会被捕捉到。3.3 线程并发下的数据安全设备数据是高频并发写入的普通HashMap肯定要出事。我几乎把所有设备状态相关的存储都用了ConcurrentHashMap数据对象设计成只读不可变的DeviceDataPointpublic class DeviceDataPoint { private final String deviceId; private final String pointCode; private final double value; private final long timestamp; // 全参数构造器 public DeviceDataPoint(String deviceId, String pointCode, double value, long timestamp) { this.deviceId deviceId; this.pointCode pointCode; this.value value; this.timestamp timestamp; } public String getDeviceId() { return deviceId; } public String getPointCode() { return pointCode; } public double getValue() { return value; } public long getTimestamp() { return timestamp; } }不可变对象在并发环境下不需要加锁随便多少个线程同时读都不会冲突。为了让GC不成为瓶颈这些对象都很小没有深层次的引用Young GC能快速回收。还有一个并发坑是在统计在线率时用Date或者LocalDateTime直接做原子更新。我在心跳检测时发现并发SimpleDateFormat会导致时间错乱最后统一改用Instant和AtomicLong存时间戳避免线程安全问题。4. 数据流转的高光时刻实时推送和WebSocket4.1 推送方案的技术取舍监控数据推送到前端大屏技术选型其实很明确WebSocket。轮询用在这里弊端很明显一秒几次每台风机都轮询接口压力太大而且近实时的体验要求也达不到。Spring Boot集成WebSocket有原生方案和封装方案我选了原生ServerEndpoint加Spring注入的方式。原生方案依赖少和Spring Boot集成起来也不复杂Component ServerEndpoint(/ws/monitor/{deviceId}) public class MonitoringWebSocketEndpoint { private static final CopyOnWriteArraySetSession sessions new CopyOnWriteArraySet(); OnOpen public void onOpen(Session session, PathParam(deviceId) String deviceId) { session.getUserProperties().put(deviceId, deviceId); sessions.add(session); } OnClose public void onClose(Session session) { sessions.remove(session); } OnError public void onError(Session session, Throwable error) { sessions.remove(session); } public static void sendToAll(String message) { for (Session session : sessions) { session.getAsyncRemote().sendText(message); } } }推送是异步方式用getAsyncRemote().sendText()而不是同步的sendText()主要是因为同步发送会阻塞调用线程如果某个前端网络慢了影响的是整个推送线程的吞吐。4.2 高并发推送时不把消息搞乱一个风场几十台风机每个WebSocket连接可能只关心某一台风机或者关心整个风场。我按deviceId对Session做分组管理推送时只给订阅了该设备的连接发消息public static void sendToDevice(String deviceId, String message) { for (Session session : sessions) { String subscribedDevice (String) session.getUserProperties().get(deviceId); if (deviceId.equals(subscribedDevice)) { session.getAsyncRemote().sendText(message); } } }前端大屏端订阅模式可以设计成两种整场总览页订阅大范围数据单机详情页订阅一台风机的所有测点。实际上每次推送的报文也不宜过大我一般把一次推送的数据压缩到几个测点一组而不是把一整个设备几十个数据点全塞一条JSON里。数据量大时JSON序列化本身也会成为CPU瓶颈。WebSocket推送给前端的数据格式大概是这样的里面同时带上了采样时间和数据质量标记{ deviceId: WF-001, pointCode: RPM, value: 1425.6, timestamp: 1735689600000, quality: GOOD }quality字段有两个值GOOD表示数据正常BAD表示设备端标记故障或者被后端的清洗逻辑打过标记。这个字段前端会根据不同颜色渲染调度员一眼就能看到哪些风机数据不可信不用等告警系统跑完一轮才反应过来。4.3 为什么不用MQTT做设备接入和前端推送项目立项时有人提议直接用MQTT把设备数据发到Broker前端订阅Broker拿数据。这个方案在公共云物联网平台上是合理的设备端SDK成熟、断线重连机制完善。但我这套项目部署在风电场内网网络环境相对可控设备数量几百台规模MQTT引入的Broker运维成本和消息流转链路延迟反而成了负担。TCP长连接加自定义协议解析逻辑自己掌控出问题能直接抓包定位。WebSocket推送给前端也是内网直连省掉一个中转环节。技术选型没有绝对的对错关键还是看部署环境和规模。5. 存储层的取舍关系库够用何必上时序库5.1 表结构设计里的降级策略我的存储方案是MySQL分表按天分表存数据。表名形如fan_data_20251220每天凌晨动态建第二天的表。这样单表数据量可控查询最近一天的数据也很快不需要动不动就上HBase或时序数据库。表结构设计上高频测点按数据点一行行存。核心字段就这几个CREATE TABLE fan_data_20251220 ( id BIGINT AUTO_INCREMENT PRIMARY KEY, device_id VARCHAR(20) NOT NULL, point_code VARCHAR(30) NOT NULL, data_value DOUBLE NOT NULL, sample_time BIGINT NOT NULL, quality_flag TINYINT DEFAULT 1, KEY idx_device_time (device_id, sample_time) ) ENGINEInnoDB;为什么不用宽表一行存所有测点因为不同设备的测点集不固定有的风机有振动测点有的老机组没有。宽表改造成本太高。竖表加point_code字段的做法扩展性最好新加一个测点不需要改表结构。代价是单条查询需要按点和时间过滤但配合索引和大屏的分页查询性能完全顶得住。数据入库我用的是分批批量插入攒够200条或500毫秒刷一次JDBC的rewriteBatchedStatementstrue参数开启后才能发挥批量插入的真实性能public void batchInsert(ListDeviceDataPoint points) { String sql INSERT INTO fan_data_20251220 (device_id, point_code, data_value, sample_time, quality_flag) VALUES (?, ?, ?, ?, ?); jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() { Override public void setValues(PreparedStatement ps, int i) throws SQLException { DeviceDataPoint point points.get(i); ps.setString(1, point.getDeviceId()); ps.setString(2, point.getPointCode()); ps.setDouble(3, point.getValue()); ps.setLong(4, point.getTimestamp()); ps.setInt(5, GOOD.equals(point.getQuality()) ? 1 : 0); } Override public int getBatchSize() { return points.size(); } }); }这样优化过之后MySQL在普通机械硬盘上也能维持每秒几千条数据的写入速度对于百台机组的风场完全够了。如果以后机组数量翻倍直接分库分表或者平滑迁移到时序库即可业务层的数据模型已经兼容了这个可能。5.2 热数据和冷数据的二分管理风机实时监控系统里工程师最常查的是最近一小时的数据其次是今天的趋势曲线一周前的数据很少点开看。这决定了我的存储策略热点数据放Redis近期数据放MySQL历史归档就做定时任务导出到文件或者迁移到汇总表。Redis里我维护每个测点最近N条数据的一个循环列表大屏的实时趋势图直接从这里取。结构上用List或者Stream的XADD都行我用的是最简单的方式public void cacheLatestData(String deviceId, String pointCode, DeviceDataPoint point) { String redisKey realtime: deviceId : pointCode; redisTemplate.opsForList().rightPush(redisKey, point); redisTemplate.opsForList().trim(redisKey, -300, -1); // 只保留最近300条 }Redis里的数据设置短TTL比如5分钟过期。大屏实时曲线正常情况下只依赖WebSocket推送的实时值历史数据回看才去查Redis或MySQL这样一套组合既保证了速度又控制了内存。6. 实测下来最值得说的经验和坑6.1 数据丢帧的真相Netty接收缓冲区溢出了系统跑了两周后我发现每天的曲线图在特定时间段会出现数据缺口查日志也没有报错。抓包后发现有的设备在突发状态下会在几百毫秒内连续发几十个报文Netty默认的接收缓冲区不够大一部分报文还没来得及读就被内核丢弃。解决方案就是调大Netty的接收缓冲区同时对设备端的发送频率做上限保护协议约定bootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(65536)); bootstrap.childOption(ChannelOption.SO_RCVBUF, 1024 * 1024);这里要注意SO_RCVBUF是内核缓冲区的建议值RCVBUF_ALLOCATOR是Netty应用层的分配器两个要配合调整只调一个未必生效。6.2 WebSocket连接被防火墙掐断的保活设计风电场的网络环境有各种安全设备长连接隔一段时间不出数据就会被防火墙当空闲连接切断。设备端到Netty网关我用的是心跳包保活但浏览器端的WebSocket连接是由Nginx反向代理的Nginx也有自己的超时机制。我针对WebSocket写了一个前端心跳let heartbeatInterval setInterval(() { if (ws.readyState WebSocket.OPEN) { ws.send(JSON.stringify({ type: PING })); } }, 30000);同时后端每收到一个PING消息就回复PONG前端连续3次没有收到PONG就主动重连。这样的机制稳定运行了几周没有再出现大屏数据断流的问题。6.3 转速和温度之间的蛛丝马迹从数据流转到联动分析把风机转速和齿轮箱温度放在一起分析能发现很多有价值的东西。比如正常情况下转速升高齿轮箱温度会在几分钟内缓慢上升如果出现转速降低但温度仍然快速上涨大概率是齿轮箱内部润滑系统出问题。我在数据流转到Service层时维护了一个设备维度的轻量状态机public void analyzeRelation(String deviceId, double rpm, double gearboxTemp) { // 转速降3%以上且温度5分钟内上涨超过8度触发疑似润滑故障 if (rpm lastRpm * 0.97 gearboxTemp lastTemp 8) { alertService.sendAlert(deviceId, GEARBOX_LUBRICATION_ABNORMAL); } lastRpm rpm; lastTemp gearboxTemp; }这个逻辑本身不算复杂但价值在于它把数据流转过程中原本孤立的测点关联了起来。实际跑下来确实抓到了一台风机早期齿轮箱异常维修队提前介入避免了更严重的设备损坏。6.4 关于Spring Boot物联网数据流转的整体心得做完这个项目我对Spring Boot在物联网场景里的定位有了更清晰的认识。Spring Boot的优势是生态完善、开发效率高事务管理、连接池、定时任务、WebSocket这些基础设施都不用重复造轮子。它不是性能瓶颈真正的瓶颈往往出在粗心大意的链路设计上。比如数据清洗环节一开始我的清洗逻辑和数据入库逻辑在同一个事务里。结果清洗逻辑慢的时候会拖慢入库网络抖动也会影响数据写入。后来干脆把清洗和入库拆开清洗完先放内存队列队列满了直接降级丢弃并打日志。业务不敏感的中间数据允许在极端情况下丢几个点好过整个服务宕机。还有数据报文解析里的位运算看起来是编码基本功却对接入层的数据准确性起着决定性作用。建议所有做设备接入的同学拿到协议文档第一件事就是把所有类型数据的字节序、符号位、缩放系数列成表格写进代码注释里甚至直接在配置中心里做成可配置项。这样设备升级、协议变更时后端只需要改配置而不用改代码。这套系统从设计到落地前后迭代了一个多月才稳定下来。现在每天稳定接收几千万条数据点大屏实时刷新延迟控制在几百毫秒内告警准确率也比以前靠人工盯数据高了很多。要说遗憾就是当初没直接在接入层把协议解析做成规则引擎目前每种设备类型还是要单独写一段解析代码。下一次迭代如果接入更多类型的传感器设备我打算用策略模式做一轮重构让新设备接入只写配置不写代码。