独立配电柜多点采集：RJ45以太网温湿度传感器，TCP粘包问题处理方案

发送端:                         接收端:
send("TEMP:25.3,HUMI:60")  ─┐
send("TEMP:26.1,HUMI:58")  ─┼─→ recv(): "TEMP:25.3,HUMI:60TEMP:26.1,HUMI:58"
send("TEMP:25.8,HUMI:61")  ─┘      ↑ 三次发送合并为一次接收（粘包）

发送端:                         接收端:
send("TEMP:25.3,HUMI:60")  ─┐→ recv(): "TEMP:25.3,H"
                             ├→ recv(): "UMI:60TEMP:26"
                             └→ recv(): ".1,HUMI:58"

┌─────────────────────────────────────────────────────────────────────┐
│ 配电柜A (1F车间)                                                   │
│                                                                     │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐                        │
│  │ 传感器#1 │  │ 传感器#2 │  │ 传感器#3 │                        │
│  │ 柜顶     │  │ 柜中     │  │ 柜底     │                        │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘                        │
│       └──────────────┼──────────────┘                            │
│                      │ RS485 (Modbus RTU)                         │
│               ┌──────▼───────┐                                    │
│               │ 以太网网关    │                                    │
│               │ (RS485→TCP)  │                                    │
│               └──────┬───────┘                                    │
│                      │ TCP Port 502                               │
└──────────────────────┼──────────────────────────────────────────────┘
                       │
┌──────────────────────┼──────────────────────────────────────────────┐
│ 园区网络              │                                            │
│                      │                                            │
│               ┌──────▼───────┐                                    │
│               │ 采集服务器    │                                    │
│               │              │                                    │
│               │  TCP客户端   │ ← 每个配电柜一个长连接              │
│               │  ┌──────────┐│                                    │
│               │  │ 粘包处理 ││ ← 核心：帧同步+缓冲区管理          │
│               │  │ 引擎     ││                                    │
│               │  └──────────┘│                                    │
│               │  ┌──────────┐│                                    │
│               │  │ 报文解析 ││ ← 按协议格式提取温湿度              │
│               │  └──────────┘│                                    │
│               │  ┌──────────┐│                                    │
│               │  │ 数据写入 ││ → InfluxDB / MySQL                 │
│               │  └──────────┘│                                    │
│               └──────────────┘                                    │
└─────────────────────────────────────────────────────────────────────┘

┌──────────┬──────────┬──────────┬──────────────────────────────────┐
│ 偏移     │ 长度     │ 字段     │ 说明                             │
├──────────┼──────────┼──────────┼──────────────────────────────────┤
│ 0        │ 2        │ 事务ID   │ 请求/响应匹配                   │
│ 2        │ 2        │ 协议ID   │ 0x0000 = Modbus                 │
│ 4        │ 2        │ 长度     │ 后续字节数（含单元ID+功能码+数据）│
│ 6        │ 1        │ 单元ID   │ 从站地址                        │
│ 7        │ 1        │ 功能码   │ 0x04 = 读输入寄存器             │
│ 8        │ N        │ 数据     │ 寄存器值                        │
└──────────┴──────────┴──────────┴──────────────────────────────────┘

// ringbuf.h
#ifndef RINGBUF_H
#define RINGBUF_H

#include <stdint.h>
#include <string.h>

#define RINGBUF_SIZE  4096

typedef struct {
    uint8_t  buf[RINGBUF_SIZE];
    uint32_t head;     // 写入位置
    uint32_t tail;     // 读取位置
    uint32_t count;    // 当前数据量
} ringbuf_t;

static inline void ringbuf_init(ringbuf_t *rb) {
    rb->head = rb->tail = rb->count = 0;
}

static inline int ringbuf_write(ringbuf_t *rb, const uint8_t *data, uint32_t len) {
    if (rb->count + len > RINGBUF_SIZE) {
        return -1;  // 缓冲区满
    }
    for (uint32_t i = 0; i < len; i++) {
        rb->buf[rb->head] = data[i];
        rb->head = (rb->head + 1) % RINGBUF_SIZE;
    }
    rb->count += len;
    return 0;
}

static inline int ringbuf_read(ringbuf_t *rb, uint8_t *data, uint32_t len) {
    if (rb->count < len) {
        return 0;  // 数据不足
    }
    for (uint32_t i = 0; i < len; i++) {
        data[i] = rb->buf[rb->tail];
        rb->tail = (rb->tail + 1) % RINGBUF_SIZE;
    }
    rb->count -= len;
    return len;
}

static inline uint32_t ringbuf_available(ringbuf_t *rb) {
    return rb->count;
}

static inline int ringbuf_peek(ringbuf_t *rb, uint8_t *data, uint32_t len) {
    if (rb->count < len) {
        return 0;
    }
    uint32_t pos = rb->tail;
    for (uint32_t i = 0; i < len; i++) {
        data[i] = rb->buf[pos];
        pos = (pos + 1) % RINGBUF_SIZE;
    }
    return len;
}

#endif

// modbus_tcp_framing.c
#include <arpa/inet.h>
#include <errno.h>
#include <netinet/in.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>
#include "ringbuf.h"

#define MODBUS_TCP_MBAP_LEN_OFFSET  4    // 长度字段在MBAP头中的偏移
#define MODBUS_TCP_MBAP_SIZE        6    // MBAP头长度（事务ID+协议ID+长度）
#define MODBUS_TCP_MIN_FRAME        8    // 最小合法帧长度
#define MODBUS_TCP_MAX_FRAME        260  // 最大合法帧长度

// Modbus TCP帧结构
#pragma pack(push, 1)
typedef struct {
    uint16_t transaction_id;   // 事务ID（大端）
    uint16_t protocol_id;      // 协议ID（0=Modbus）
    uint16_t length;           // 后续字节数（大端）
    uint8_t  unit_id;          // 单元ID
    uint8_t  function_code;    // 功能码
    uint8_t  data[];           // 数据（变长）
} modbus_tcp_frame_t;
#pragma pack(pop)

// 解析状态
typedef enum {
    STATE_WAIT_HEADER,     // 等待MBAP头（6字节）
    STATE_WAIT_PAYLOAD,    // 等待载荷数据
    STATE_FRAME_COMPLETE,  // 完整帧已就绪
    STATE_ERROR,           // 解析错误
} parse_state_t;

// 粘包解析器
typedef struct {
    ringbuf_t       rb;           // 接收环形缓冲区
    parse_state_t   state;
    uint8_t         header_buf[MODBUS_TCP_MBAP_SIZE];
    uint8_t        *payload_buf;
    uint16_t        expected_payload_len;  // 期望的载荷长度
    uint16_t        received_payload_len;  // 已接收的载荷长度
    uint16_t        transaction_id;        // 当前帧的事务ID
} framer_t;

static framer_t g_framer;

// 初始化解析器
void framer_init(framer_t *f) {
    ringbuf_init(&f->rb);
    f->state = STATE_WAIT_HEADER;
    f->payload_buf = NULL;
    f->expected_payload_len = 0;
    f->received_payload_len = 0;
}

// 从缓冲区中尝试提取一个完整帧
// 返回: 0=需要更多数据, >0=完整帧长度, <0=错误
int framer_extract_frame(framer_t *f, uint8_t *out_frame, uint32_t max_len) {
    while (1) {
        switch (f->state) {

        case STATE_WAIT_HEADER: {
            // 需要至少6字节的MBAP头
            if (ringbuf_available(&f->rb) < MODBUS_TCP_MBAP_SIZE) {
                return 0;  // 数据不足，等待更多
            }

            // 读取MBAP头
            ringbuf_read(&f->rb, f->header_buf, MODBUS_TCP_MBAP_SIZE);

            // 验证协议ID（应为0）
            uint16_t proto_id = (f->header_buf[2] << 8) | f->header_buf[3];
            if (proto_id != 0x0000) {
                // 协议ID不对，丢弃这6个字节，尝试重新同步
                f->state = STATE_WAIT_HEADER;
                continue;
            }

            // 提取长度字段（大端）
            f->expected_payload_len = (f->header_buf[4] << 8) | f->header_buf[5];

            // 合法性检查
            if (f->expected_payload_len < 2 || f->expected_payload_len > 254) {
                // 长度异常，可能是粘包导致的错位
                // 丢弃已读的头，继续尝试
                f->state = STATE_WAIT_HEADER;
                continue;
            }

            // 分配载荷缓冲区
            if (f->payload_buf) {
                free(f->payload_buf);
            }
            f->payload_buf = malloc(f->expected_payload_len);
            if (!f->payload_buf) {
                f->state = STATE_ERROR;
                return -1;
            }

            f->received_payload_len = 0;
            f->state = STATE_WAIT_PAYLOAD;
            // 不break，直接进入PAYLOAD状态尝试读取
        }
        /* fall through */

        case STATE_WAIT_PAYLOAD: {
            uint32_t avail = ringbuf_available(&f->rb);
            uint32_t need = f->expected_payload_len - f->received_payload_len;

            if (avail == 0) {
                return 0;  // 等待更多数据
            }

            // 读取尽可能多的数据
            uint32_t to_read = (avail < need) ? avail : need;
            ringbuf_read(&f->rb, f->payload_buf + f->received_payload_len, to_read);
            f->received_payload_len += to_read;

            if (f->received_payload_len >= f->expected_payload_len) {
                // 完整帧已就绪
                f->state = STATE_FRAME_COMPLETE;
                // 不break，继续到COMPLETE状态
            } else {
                return 0;  // 还需要更多数据
            }
        }
        /* fall through */

        case STATE_FRAME_COMPLETE: {
            // 组装完整帧：MBAP头 + 载荷
            uint32_t total_len = MODBUS_TCP_MBAP_SIZE + f->expected_payload_len;
            if (total_len > max_len) {
                f->state = STATE_ERROR;
                return -1;
            }

            memcpy(out_frame, f->header_buf, MODBUS_TCP_MBAP_SIZE);
            memcpy(out_frame + MODBUS_TCP_MBAP_SIZE, f->payload_buf, f->expected_payload_len);

            // 提取事务ID
            f->transaction_id = (f->header_buf[0] << 8) | f->header_buf[1];

            // 重置状态，准备下一帧
            free(f->payload_buf);
            f->payload_buf = NULL;
            f->state = STATE_WAIT_HEADER;

            return total_len;
        }

        case STATE_ERROR:
        default:
            return -1;
        }
    }
}

// 从socket读取数据并放入缓冲区
int framer_feed_socket(framer_t *f, int sockfd) {
    uint8_t raw_buf[1024];
    ssize_t n = recv(sockfd, raw_buf, sizeof(raw_buf), MSG_DONTWAIT);
    if (n <= 0) {
        if (n == 0) {
            return -1;  // 连接关闭
        }
        if (errno == EAGAIN || errno == EWOULDBLOCK) {
            return 0;  // 暂无数据
        }
        return -1;  // 错误
    }

    if (ringbuf_write(&f->rb, raw_buf, n) != 0) {
        return -1;  // 缓冲区满
    }

    return n;
}

// collector.c
#include <arpa/inet.h>
#include <netinet/in.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>
#include "modbus_tcp_framing.h"

#define SERVER_IP      "10.90.1.50"
#define SERVER_PORT    502
#define POLL_INTERVAL  10    // 轮询间隔（秒）
#define RECV_TIMEOUT   3     // 接收超时（秒）

// 传感器配置
typedef struct {
    uint8_t  unit_id;       // Modbus从站地址
    uint16_t temp_reg;      // 温度寄存器地址
    uint16_t humi_reg;      // 湿度寄存器地址
    float    temp;          // 最新温度
    float    humi;          // 最新湿度
} sensor_cfg_t;

static sensor_cfg_t g_sensors[4] = {
    {1, 0x0000, 0x0001, 0, 0},  // 传感器#1，柜顶
    {2, 0x0000, 0x0001, 0, 0},  // 传感器#2，柜中
    {3, 0x0000, 0x0001, 0, 0},  // 传感器#3，柜底
    {4, 0x0000, 0x0001, 0, 0},  // 传感器#4，电缆口
};

static int g_sockfd = -1;
static pthread_mutex_t g_sock_mutex = PTHREAD_MUTEX_INITIALIZER;
static uint16_t g_transaction_id = 0;

// 构建Modbus TCP读寄存器请求
static int build_read_request(uint8_t *buf, uint16_t trans_id,
                               uint8_t unit_id, uint16_t reg_addr,
                               uint16_t reg_count) {
    uint16_t *p = (uint16_t *)buf;

    p[0] = htons(trans_id);        // 事务ID
    p[1] = htons(0x0000);          // 协议ID
    p[2] = htons(6);               // 长度（后续6字节）
    buf[6] = unit_id;              // 单元ID
    buf[7] = 0x04;                 // 功能码：读输入寄存器
    p[4] = htons(reg_addr);        // 起始寄存器地址
    p[5] = htons(reg_count);       // 寄存器数量

    return 12;  // Modbus TCP请求固定12字节
}

// 解析Modbus TCP响应中的温湿度数据
static int parse_response(uint8_t *frame, uint32_t len,
                           float *temp, float *humi) {
    if (len < MODBUS_TCP_MBAP_SIZE + 5) {
        return -1;  // 帧太短
    }

    modbus_tcp_frame_t *resp = (modbus_tcp_frame_t *)frame;
    uint8_t byte_count = resp->data[0];  // 数据字节数

    if (byte_count < 4) {
        return -1;  // 至少需要4字节（2个寄存器）
    }

    // 温度：第1个寄存器，0.1℃为单位
    int16_t raw_temp = (resp->data[1] << 8) | resp->data[2];
    *temp = raw_temp / 10.0f;

    // 湿度：第2个寄存器，0.1%RH为单位
    int16_t raw_humi = (resp->data[3] << 8) | resp->data[4];
    *humi = raw_humi / 10.0f;

    return 0;
}

// 轮询单个传感器
static int poll_sensor(sensor_cfg_t *sensor, framer_t *framer) {
    uint8_t req_buf[12];
    uint8_t resp_buf[256];
    int req_len;

    // 构建请求
    g_transaction_id++;
    req_len = build_read_request(req_buf, g_transaction_id,
                                  sensor->unit_id,
                                  sensor->temp_reg, 2);

    // 发送请求
    pthread_mutex_lock(&g_sock_mutex);
    ssize_t sent = send(g_sockfd, req_buf, req_len, 0);
    if (sent != req_len) {
        pthread_mutex_unlock(&g_sock_mutex);
        return -1;
    }

    // 等待响应（带超时）
    struct timeval tv_start, tv_now;
    gettimeofday(&tv_start, NULL);

    while (1) {
        // 从socket读取数据到framer缓冲区
        framer_feed_socket(framer, g_sockfd);

        // 尝试提取完整帧
        int frame_len = framer_extract_frame(framer, resp_buf, sizeof(resp_buf));
        if (frame_len > 0) {
            // 验证事务ID
            uint16_t resp_tid = (resp_buf[0] << 8) | resp_buf[1];
            if (resp_tid == g_transaction_id) {
                // 解析数据
                float temp, humi;
                if (parse_response(resp_buf, frame_len, &temp, &humi) == 0) {
                    sensor->temp = temp;
                    sensor->humi = humi;
                    pthread_mutex_unlock(&g_sock_mutex);
                    return 0;
                }
            }
            // 事务ID不匹配或解析失败，继续等待
        }

        // 检查超时
        gettimeofday(&tv_now, NULL);
        double elapsed = (tv_now.tv_sec - tv_start.tv_sec) +
                         (tv_now.tv_usec - tv_start.tv_usec) / 1e6;
        if (elapsed > RECV_TIMEOUT) {
            pthread_mutex_unlock(&g_sock_mutex);
            return -1;  // 超时
        }

        usleep(10000);  // 10ms
    }
}

// 连接服务器
static int connect_server(void) {
    struct sockaddr_in servaddr;

    if (g_sockfd >= 0) {
        close(g_sockfd);
    }

    if ((g_sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
        perror("socket creation failed");
        return -1;
    }

    // 设置TCP_NODELAY（禁用Nagle算法）
    int flag = 1;
    setsockopt(g_sockfd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag));

    // 设置接收超时
    struct timeval tv;
    tv.tv_sec = RECV_TIMEOUT;
    tv.tv_usec = 0;
    setsockopt(g_sockfd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));

    memset(&servaddr, 0, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_port = htons(SERVER_PORT);
    servaddr.sin_addr.s_addr = inet_addr(SERVER_IP);

    if (connect(g_sockfd, (struct sockaddr *)&servaddr, sizeof(servaddr)) < 0) {
        perror("connect failed");
        close(g_sockfd);
        g_sockfd = -1;
        return -1;
    }

    printf("Connected to %s:%d\n", SERVER_IP, SERVER_PORT);
    return 0;
}

int main(int argc, char *argv[]) {
    framer_t framer;
    int consecutive_errors = 0;

    framer_init(&framer);

    while (1) {
        // 检查连接状态
        if (g_sockfd < 0) {
            if (connect_server() < 0) {
                sleep(5);
                continue;
            }
        }

        // 轮询所有传感器
        for (int i = 0; i < 4; i++) {
            int rc = poll_sensor(&g_sensors[i], &framer);
            if (rc == 0) {
                printf("Sensor #%d: T=%.1fC H=%.1f%%\n",
                       i + 1, g_sensors[i].temp, g_sensors[i].humi);
                consecutive_errors = 0;
            } else {
                printf("Sensor #%d: poll failed\n", i + 1);
                consecutive_errors++;
            }
        }

        // 连续错误过多，重连
        if (consecutive_errors >= 10) {
            printf("Too many errors, reconnecting...\n");
            close(g_sockfd);
            g_sockfd = -1;
            consecutive_errors = 0;
            framer_init(&framer);  // 重置解析器
            sleep(5);
            continue;
        }

        sleep(POLL_INTERVAL);
    }

    return 0;
}

// 禁用Nagle算法（关键！）
int flag = 1;
setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag));

// 设置SO_RCVBUF（接收缓冲区）
int rcvbuf = 256 * 1024;  // 256KB
setsockopt(sockfd, SOL_SOCKET, SO_RCVBUF, &rcvbuf, sizeof(rcvbuf));

// 设置TCP_KEEPALIVE（检测死连接）
int keepalive = 1;
setsockopt(sockfd, SOL_SOCKET, SO_KEEPALIVE, &keepalive, sizeof(keepalive));

int keepidle = 60;    // 60秒无数据后开始探测
int keepintvl = 10;   // 探测间隔10秒
int keepcnt = 3;      // 3次失败断开
setsockopt(sockfd, IPPROTO_TCP, TCP_KEEPIDLE, &keepidle, sizeof(keepidle));
setsockopt(sockfd, IPPROTO_TCP, TCP_KEEPINTVL, &keepintvl, sizeof(keepintvl));
setsockopt(sockfd, IPPROTO_TCP, TCP_KEEPCNT, &keepcnt, sizeof(keepcnt));

// 发送请求时记录事务ID
uint16_t expected_tid = g_transaction_id;
send_request(expected_tid, ...);

// 接收响应时验证
while (1) {
    frame = framer_extract_frame(...);
    uint16_t resp_tid = extract_tid(frame);
    if (resp_tid == expected_tid) {
        // 这是我要的响应
        break;
    }
    // 不是我的响应，可能是粘包中前一个请求的残留
    // 继续提取下一帧
}

┌──────────┐     ┌──────────┐     ┌──────────┐
│ 发送请求 │────▶│ 等待响应 │────▶│ 超时?    │
│          │     │ (3秒)    │     │          │
└──────────┘     └──────────┘     └───┬───┬──┘
                                      │   │
                              否      │   │ 是
                              ◀───────┘   └──▶ 重传(最多3次)
                                                    │
                                          仍失败 ────┤
                                                    ▼
                                              标记设备离线
                                              触发告警

// collector.go
package main

import (
	"bufio"
	"encoding/binary"
	"fmt"
	"net"
	"time"
)

const (
	serverIP   = "10.90.1.50"
	serverPort = 502
)

type ModbusTCPHeader struct {
	TransactionID uint16
	ProtocolID    uint16
	Length        uint16
	UnitID        uint8
	FunctionCode  uint8
}

type Collector struct {
	conn   net.Conn
	reader *bufio.Reader
}

func (c *Collector) readFrame() ([]byte, error) {
	// 先读6字节MBAP头
	header := make([]byte, 6)
	if _, err := c.reader.Read(header); err != nil {
		return nil, err
	}

	// 解析长度字段
	length := binary.BigEndian.Uint16(header[4:6])

	// 读取剩余数据
	payload := make([]byte, length)
	if _, err := c.reader.Read(payload); err != nil {
		return nil, err
	}

	// 组装完整帧
	frame := append(header, payload...)
	return frame, nil
}

func (c *Collector) pollSensor(unitID byte, regAddr, regCount uint16) (float32, float32, error) {
	// 构建请求
	transID := uint16(time.Now().UnixNano() & 0xFFFF)
	req := buildRequest(transID, unitID, regAddr, regCount)

	// 发送
	if _, err := c.conn.Write(req); err != nil {
		return 0, 0, err
	}

	// 读取响应
	frame, err := c.readFrame()
	if err != nil {
		return 0, 0, err
	}

	// 验证事务ID
	respTID := binary.BigEndian.Uint16(frame[0:2])
	if respTID != transID {
		return 0, 0, fmt.Errorf("transaction ID mismatch")
	}

	// 解析温湿度
	return parseData(frame[7:])
}

func main() {
	conn, err := net.DialTimeout("tcp", fmt.Sprintf("%s:%d", serverIP, serverPort), 5*time.Second)
	if err != nil {
		panic(err)
	}
	defer conn.Close()

	// 禁用Nagle
	if tcpConn, ok := conn.(*net.TCPConn); ok {
		tcpConn.SetNoDelay(true)
	}

	collector := &Collector{
		conn:   conn,
		reader: bufio.NewReaderSize(conn, 4096),
	}

	for {
		temp, humi, err := collector.pollSensor(1, 0x0000, 2)
		if err != nil {
			fmt.Printf("Poll failed: %v\n", err)
		} else {
			fmt.Printf("T=%.1fC H=%.1f%%\n", temp, humi)
		}
		time.Sleep(10 * time.Second)
	}
}

# 抓包查看Modbus TCP通信
tcpdump -i eth0 -A -s 0 'port 502'

# 使用modbus-cli测试
modbus tcp read --host 10.90.1.50 --port 502 --unit-id 1 \
  --register-type input --address 0 --count 2

# 查看TCP连接状态
ss -tnp | grep 502

# 查看TCP重传统计
cat /proc/net/netstat | grep -i retrans