TCP/UDP 私有协议

名尚智联通过自研 Netty 多协议接入框架（ms-netty-core），同时监听 TCP 11000 和 UDP 11001，支持自定义二进制协议。

---

一、MS-PRIVATE v1 协议

1.1 帧格式

+--------+--------+-----------+------------+--------+
| 魔数 4B | 版本 1B | 类型 1B  | 序列号 2B  | 载荷   |
+--------+--------+-----------+------------+--------+
| 0x4D530101 | 0x01 | 0-6    | 自增       | JSON   |
+--------+--------+-----------+------------+--------+

类型	值	方向	说明
HELLO	0x00	C→S	握手（serial + authCode）
HELLO_ACK	0x01	S→C	握手响应
PROPS	0x02	C→S	属性上报
PROPS_ACK	0x03	S→C	属性响应
PING	0x04	C→S	心跳
PONG	0x05	S→C	心跳响应
COMMAND	0x06	S→C	平台下发指令

1.2 载荷示例

HELLO：{"serial":"DEVICE-001","auth":"a1b2c3..."}PROPS：{"props":{"temperature":25.6}}COMMAND：{"cmd":"restart","id":"xxx"}

---

二、Netty Pipeline

ByteToFrameDecoder        ← 4B 魔数 + payload length 解码
  → ProtocolHandler      ← 类型分发（HELLO/PROPS/COMMAND）
    → DeviceAuthenticator ← serial + authCode 校验（主库 + tenant_* 遍历）
      → DeviceDataDispatcher ← 写 Redis TSLV + TDengine

2.1 PipelineProvider SPI

各协议模块通过 UdpPipelineProvider / TcpPipelineProvider 自注册：

public interface TcpPipelineProvider {
    String protocol();     // "ms-private"
    int port();            // 0=用全局 11000，>0=独立端口
    ChannelInitializer initializer();
}

• ms.netty.tcp.enabled / ms.netty.udp.enabled 全局门控
• Provider.port() 区分独立端口（如 CoAP 5683 独立于 UDP 11001）

---

三、认证流程

设备 → TCP 11000 建立连接
  → 发送 HELLO 帧（serial + authCode）
    → 主库 ms_device 查 serial
      → 未找到 → tenant_* 动态库遍历（TreeMap 排序）
        → 找到 → authCode 明文比对
          → 匹配 → HELLO_ACK + 会话注册（TTL 10 分钟）
          → 不匹配 → 断开
        → 未找到 → 断开
      → 找到 → authCode 比对（同上）
  → 后续帧（PROPS/PING）复用会话，无需重复认证

核心工具方法：

DeviceMapper.selectBySerialIgnoreTenant(@InterceptorIgnore tenantLine);
// 主库查不到时，遍历 DynamicRoutingDataSource 的所有 tenant_* 数据源

---

四、Python 客户端示例

import socket
import struct
import json
import threading
import time

MAGIC = 0x4D530101
VERSION = 0x01

def build_frame(msg_type: int, seq: int, payload: dict) -> bytes:
    header = struct.pack('>IBBH', MAGIC, VERSION, msg_type, seq)
    body = json.dumps(payload).encode('utf-8')
    return header + body

def parse_frame(data: bytes):
    magic, ver, typ, seq = struct.unpack('>IBBH', data[:12])
    body = json.loads(data[12:].decode('utf-8')) if len(data) > 12 else {}
    return magic, ver, typ, seq, body

class PrivateDevice:
    MSG_HELLO = 0x00
    MSG_HELLO_ACK = 0x01
    MSG_PROPS = 0x02
    MSG_PING = 0x04
    MSG_PONG = 0x05
    MSG_COMMAND = 0x06

    def __init__(self, host, port, serial, auth):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.connect((host, port))
        self.serial = serial
        self.auth = auth
        self.seq = 0

    def recv_loop(self):
        while True:
            try:
                data = self.sock.recv(4096)
                if data:
                    magic, ver, typ, seq, body = parse_frame(data)
                    if typ == self.MSG_COMMAND:
                        print(f"[cmd] {body}")
                    elif typ == self.MSG_HELLO_ACK:
                        print(f"[hello-ack] ok")
            except: break

    def hello(self):
        seq = next_seq()
        self.sock.send(build_frame(self.MSG_HELLO, seq,
                        {"serial": self.serial, "auth": self.auth}))

    def report(self, props: dict):
        seq = next_seq()
        self.sock.send(build_frame(self.MSG_PROPS, seq, {"props": props}))

    def ping(self):
        seq = next_seq()
        self.sock.send(build_frame(self.MSG_PING, seq, {}))

    def close(self): self.sock.close()

    def next_seq(self):
        self.seq = (self.seq + 1) % 65536
        return self.seq

# 使用
dev = PrivateDevice("your-server.com", 11000, "DEVICE-001", "a1b2c3...")
threading.Thread(target=dev.recv_loop, daemon=True).start()
dev.hello()

while True:
    dev.report({"temperature": 25.6, "humidity": 62.3})
    dev.ping()
    time.sleep(10)

---

五、服务端配置

ms:
  netty:
    tcp:
      enabled: true
      port: 11000
      boss-threads: 1
      worker-threads: 0        # 0=CPU 核数 × 2
      reader-idle-seconds: 300 # 超过则断开离线连接
    udp:
      enabled: true
      port: 11001
      broadcast: true

---

六、与 MQTT 对比

维度	TCP/UDP 私有协议	MQTT
开销	极低（二进制帧）	中（TCP + MQTT 头）
QoS	自研（可选）	0/1/2
传输	TCP + UDP	仅 TCP（WS 也是 TCP）
实现成本	需写客户端	用现成 SDK
适用场景	资源极受限设备、自定义时序	通用 IoT