工厂环境监测系统实战（二）：TCP 传感器数据采集与解析

前言上一章介绍了系统整体架构本章介绍数据采集层——如何通过 TCP 协议接收传感器设备上传的环境数据并实时解析入库。为什么用 TCP工业传感器通常通过串口RS485/RS232或以太网传输数据。本系统中传感器设备通过以太网以 TCP 方式主动上报 JSON 格式数据服务端只需监听端口接收即可。TCP 相比 HTTP 延迟更低适合持续高频的数据上报。监听服务实现TcpSensorService继承自BackgroundService随 .NET Core 应用启动自动运行public class TcpSensorService : BackgroundService { private readonly ILoggerTcpSensorService _logger; private readonly int _port 9001; protected override async Task ExecuteAsync(CancellationToken stoppingToken) { var listener new TcpListener(IPAddress.Any, _port); listener.Start(); _logger.LogInformation($[TCP传感器] 监听端口 {_port} 已启动); while (!stoppingToken.IsCancellationRequested) { var client await listener.AcceptTcpClientAsync(); _logger.LogInformation($[TCP传感器] 客户端连接: {client.Client.RemoteEndPoint}); _ Task.Run(() HandleClient(client, stoppingToken), stoppingToken); } } }每个传感器连接进来时新开一个 Task 独立处理不阻塞其他设备。数据接收与粘包处理TCP 是流式协议存在粘包问题——多条 JSON 可能连在一起发过来。解决方案是先按}{切分private async Task HandleClient(TcpClient client, CancellationToken stoppingToken) { var buffer new byte[4096]; var recvBuf new StringBuilder(); using (client) { var stream client.GetStream(); while (!stoppingToken.IsCancellationRequested client.Connected) { int bytesRead await stream.ReadAsync(buffer, 0, buffer.Length, stoppingToken); if (bytesRead 0) break; var chunk Encoding.UTF8.GetString(buffer, 0, bytesRead); recvBuf.Append(chunk); // 粘包处理在 }{ 之间插入换行符 var data recvBuf.ToString().Replace(}{, }\n{); var lines data.Split(\n); recvBuf.Clear(); foreach (var l in lines) ProcessLine(l.Trim()); } } }JSON 解析传感器数据可能是单条 JSON 对象也可能是数组private void ProcessLine(string line) { if (string.IsNullOrEmpty(line)) return; ListEnviromentDataViewModel items; if (line.StartsWith([)) items JsonConvert.DeserializeObjectListEnviromentDataViewModel(line); else if (line.StartsWith({)) items new ListEnviromentDataViewModel { JsonConvert.DeserializeObjectEnviromentDataViewModel(line) }; else return; foreach (var item in items) { // BN 设备编号, DT 数据类型(1温度/2湿度/3露点), V 数值 // 存入待发送队列 ... } }批量入库传感器上报频率很高每2-3秒一条如果每条都立即写数据库会造成大量 IO。这里采用了500ms 批次窗口的策略private void ResetFlushTimer(string bn) { // 取消旧定时器 if (_flushTimers.TryRemove(bn, out var oldCts)) oldCts.Cancel(); var cts new CancellationTokenSource(); _flushTimers[bn] cts; _ Task.Run(async () { await Task.Delay(BatchWindow, cts.Token); // 500ms 窗口 if (!cts.Token.IsCancellationRequested) FlushAndSend(bn); }, cts.Token); }每个设备独立一个定时器——有新数据到达就重置计时500ms 内没有新数据到达则批量入库。频控防抖同一个设备同一种数据类型如 BN1, DT1至少间隔 1 秒才处理一条避免重复入库if (_lastSendTime.TryGetValue(key, out var lastTime)) if (DateTime.Now - lastTime MinSendInterval) return; // 跳过小结本章实现了 TCP 传感器数据的高效接收与批量入库核心要点BackgroundService实现常驻后台监听粘包处理}{→}\n{500ms 批次窗口减少数据库写入频率频控防抖避免重复数据 上一篇第1章 — 项目概述与架构设计 下一篇第3章 — 数据库设计与 Entity Framework 实战