蜗牛java 知识库

Netty接入SpringBoot-iot-gateway实现

Netty 接入 SpringBoot 实现文档（iot-gateway-service）

1. 项目结构（iot-gateway-service）

iot-gateway-service/
├── pom.xml
└── src/main/java/com/example/iotgateway/
    ├── IotGatewayApplication.java
    ├── config/
    │   └── NettyServerConfig.java          # Netty 服务端配置与启动
    ├── protocol/
    │   ├── DeviceMessage.java              # 报文头 + 报文体 POJO
    │   ├── MessageType.java                # 报文类型枚举（1心跳 2遥测...）
    │   └── ProtocolCodec.java              # 自定义编解码（魔数/版本/长度校验）
    ├── netty/
    │   ├── AuthHandler.java                # 首包鉴权
    │   ├── HeartbeatHandler.java           # 心跳 + 离线检测
    │   ├── DeviceMsgHandler.java           # 业务分发（提交线程池）
    │   └── ChannelManager.java             # 在线连接管理（Channel 与设备映射）
    ├── service/
    │   ├── TelemetryWriteService.java      # 写 InfluxDB3
    │   ├── DeviceStatusService.java        # 更新 MySQL + Redis 设备状态
    │   └── EventPublishService.java        # 发 RocketMQ
    └── config/
        └── BizThreadPoolConfig.java        # 业务线程池

2. pom.xml 依赖

<dependencies>
    <!-- Spring Boot Web（本项目只需 Web 之外的 Netty，若不需要 HTTP 可不加） -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>

    <!-- Netty 4.1 -->
    <dependency>
        <groupId>io.netty</groupId>
        <artifactId>netty-all</artifactId>
        <version>4.1.115.Final</version>
    </dependency>

    <!-- 配置中心 + 注册中心 -->
    <dependency>
        <groupId>com.alibaba.cloud</groupId>
        <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
    </dependency>

    <!-- MySQL + MyBatis-Plus -->
    <dependency>
        <groupId>com.baomidou</groupId>
        <artifactId>mybatis-plus-spring-boot3-starter</artifactId>
        <version>3.5.7</version>
    </dependency>

    <!-- InfluxDB 3 Core 客户端（注意：不是 2.x 的 influxdb-client） -->
    <dependency>
        <groupId>io.influxdb</groupId>
        <artifactId>influxdb3-client-java</artifactId>
        <version>0.9.0</version>
    </dependency>

    <!-- Redis -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>

    <!-- RocketMQ -->
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.3.0</version>
    </dependency>
</dependencies>

⚠️ 最容易踩的坑：InfluxDB 依赖不要用 com.influxdb:influxdb-client-java（那是 2.x 的），本项目要用 io.influxdb:influxdb3-client-java。

3. 协议定义

3.1 报文结构（20 字节头 + JSON 体）

字段	长度	说明
魔数 Magic	2B	`0xDCDC`
版本 Version	1B	`0x01`
报文类型 Type	1B	1心跳 2遥测 3状态 4告警 5换电进度 6鉴权
设备 ID DeviceId	12B	ASCII，右侧补空格
体长度 Length	4B	报文体 JSON 字节数

3.2 POJO

// MessageType.java
public enum MessageType {
    HEARTBEAT(1), TELEMETRY(2), STATUS(3), ALARM(4), PROGRESS(5), AUTH(6);
    private final int code;
    MessageType(int code) { this.code = code; }
    public static MessageType fromCode(int code) {
        for (MessageType t : values()) if (t.code == code) return t;
        throw new IllegalArgumentException("未知报文类型: " + code);
    }
}

// DeviceMessage.java —— 解析后的完整报文
@Data
public class DeviceMessage {
    private int magic;            // 0xDCDC
    private int version;          // 0x01
    private MessageType type;     // 报文类型
    private String deviceId;      // 设备ID（去空格）
    private String bodyJson;      // 报文体 JSON 原串
    // 可选：解析 bodyJson 后的对象（用 JsonNode 或具体 DTO）
    private Map<String, Object> body;  // 反序列化后的内容
}

4. 解码器（ProtocolCodec）

负责：校验魔数/版本 → 按长度切出 body → 组装 DeviceMessage。

public class ProtocolCodec extends ByteToMessageDecoder {
    private static final int HEADER_LENGTH = 20;
    private static final short MAGIC = (short) 0xDCDC;

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        if (in.readableBytes() < HEADER_LENGTH) return;  // 头未齐，等下一批

        short magic = in.getShort(0);
        if (magic != MAGIC) {
            // 魔数非法：计数，连续 10 次断开连接（BR-011）
            ctx.close();
            return;
        }
        byte version = in.getByte(2);
        int type = in.getByte(3) & 0xFF;
        // 设备ID：偏移 4，12 字节
        byte[] idBytes = new byte[12];
        in.getBytes(4, idBytes);
        String deviceId = new String(idBytes, StandardCharsets.US_ASCII).trim();
        int length = in.getInt(16);          // 体长度
        if (in.readableBytes() < HEADER_LENGTH + length) return;  // 体未齐

        // 读取完整报文
        byte[] all = new byte[HEADER_LENGTH + length];
        in.readBytes(all);
        String bodyJson = new String(all, HEADER_LENGTH, length, StandardCharsets.UTF_8);

        DeviceMessage msg = new DeviceMessage();
        msg.setMagic(magic);
        msg.setVersion(version);
        msg.setType(MessageType.fromCode(type));
        msg.setDeviceId(deviceId);
        msg.setBodyJson(bodyJson);
        out.add(msg);   // 传给下一个 Handler
    }
}

关键：先用 get* 检查头部字段、readableBytes() 判断是否收够，收不齐就 return 等下一批——配合 LengthFieldBasedFrameDecoder 你甚至可以不用手写切包，只在收到完整包后做解析。

5. Netty 服务端配置（NettyServerConfig）

@Configuration
public class NettyServerConfig {
    @Value("${iot.gateway.port:9000}")
    private int port;

    @Resource
    private ProtocolCodec codec;
    @Resource
    private AuthHandler authHandler;
    @Resource
    private HeartbeatHandler heartbeatHandler;
    @Resource
    private DeviceMsgHandler deviceMsgHandler;

    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;

    @PostConstruct
    public void start() throws InterruptedException {
        bossGroup = new NioEventLoopGroup(1);
        workerGroup = new NioEventLoopGroup(4);
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
         .channel(NioServerSocketChannel.class)
         .option(ChannelOption.SO_BACKLOG, 1024)
         .childOption(ChannelOption.TCP_NODELAY, true)
         .childOption(ChannelOption.SO_KEEPALIVE, true)
         .childHandler(new ChannelInitializer<SocketChannel>() {
             @Override protected void initChannel(SocketChannel ch) {
                 ch.pipeline().addLast(
                     new LengthFieldBasedFrameDecoder(65535, 16, 4, 0, 0), // 解粘包
                     codec,               // 协议解析（收到完整包后触发）
                     authHandler,         // 首包鉴权，通过后移除
                     heartbeatHandler,    // 心跳/空闲检测
                     deviceMsgHandler     // 业务分发
                 );
             }
         });
        b.bind(port).sync();
        log.info("iot-gateway Netty 启动，端口 {}", port);
    }

    @PreDestroy
    public void stop() {
        if (workerGroup != null) workerGroup.shutdownGracefully();
        if (bossGroup != null) bossGroup.shutdownGracefully();
        log.info("iot-gateway Netty 已优雅关闭");
    }
}

注意：所有自定义 Handler 通过 @Resource 注入，说明它们是 Spring Bean（单例、无状态），由多个 Channel 共享。Handler 内不能存每连接的状态，连接维度的状态放 Channel 的 attr 或独立的 ChannelManager。

6. 鉴权 Handler（AuthHandler）

@Component
@Slf4j
public class AuthHandler extends ChannelInboundHandlerAdapter {
    @Resource
    private DeviceAuthService deviceAuthService;  // 查 MySQL t_device 校验 deviceToken

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        DeviceMessage m = (DeviceMessage) msg;
        if (m.getType() == MessageType.AUTH) {
            String deviceId = m.getDeviceId();
            String token = (String) m.getBody().get("deviceToken");
            if (deviceAuthService.verify(deviceId, token)) {
                // 鉴权通过：登记连接，移除本 Handler
                ChannelManager.add(deviceId, ctx.channel());
                ctx.channel().attr(AttributeKey.valueOf("deviceId")).set(deviceId);
                ctx.pipeline().remove(this);
                log.info("设备鉴权通过: {}", deviceId);
            } else {
                log.warn("设备鉴权失败: {}", deviceId);
                ctx.close();   // 3 秒内断连（BR-010）
            }
        } else {
            // 首包不是鉴权 → 拒绝
            ctx.close();
        }
    }
}

7. 心跳 / 离线检测（HeartbeatHandler）

@Component
@Slf4j
public class HeartbeatHandler extends ChannelInboundHandlerAdapter {
    @Resource
    private DeviceStatusService deviceStatusService;

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof IdleStateEvent) {
            // 读空闲超时 → 判定离线（BR-012），发 device.offline
            String deviceId = ctx.channel().attr(AttributeKey.valueOf("deviceId")).get();
            log.warn("设备心跳超时离线: {}", deviceId);
            deviceStatusService.markOffline(deviceId);
            EventPublishService.publishOffline(deviceId);
            ctx.close();
        } else {
            ctx.fireUserEventTriggered(evt);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        DeviceMessage m = (DeviceMessage) msg;
        if (m.getType() == MessageType.HEARTBEAT) {
            deviceStatusService.refreshHeartbeat(m.getDeviceId()); // Redis + MySQL 更新在线
        } else {
            ctx.fireChannelRead(msg);   // 非心跳继续往下传
        }
    }
}

8. 业务分发 Handler（DeviceMsgHandler）

核心：接收报文 → 提交业务线程池 → 按类型分流。绝不能直接在 Netty 线程里做 IO。

@Component
@Slf4j
public class DeviceMsgHandler extends SimpleChannelInboundHandler<DeviceMessage> {
    @Resource
    private BizThreadPoolConfig bizPool;          // 业务线程池
    @Resource
    private TelemetryWriteService telemetryWrite; // 写 InfluxDB3
    @Resource
    private DeviceStatusService deviceStatus;     // 写 MySQL + Redis
    @Resource
    private EventPublishService eventPublish;     // 发 RocketMQ

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, DeviceMessage m) {
        bizPool.execute(() -> {
            try {
                switch (m.getType()) {
                    case TELEMETRY -> telemetryWrite.write(m);          // 遥测 → InfluxDB3
                    case STATUS    -> deviceStatus.update(m);           // 状态 → MySQL + Redis，发 device.status
                    case ALARM     -> { deviceStatus.handleAlarm(m); eventPublish.alarm(m); }  // 告警 → 发 device.alarm
                    case PROGRESS  -> eventPublish.progress(m);         // 换电进度 → 发 swap.progress
                    default        -> log.debug("忽略类型: {}", m.getType());
                }
            } catch (Exception e) {
                log.error("处理报文失败 deviceId={}", m.getDeviceId(), e);
            }
        });
    }
}

业务线程池（BizThreadPoolConfig）：

@Configuration
public class BizThreadPoolConfig {
    @Bean("bizPool")
    public ThreadPoolExecutor bizPool() {
        return new ThreadPoolExecutor(
            8, 16, 60, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(10000),
            Thread.ofVirtual()/* 或自定义命名工厂 */,
            new ThreadPoolExecutor.CallerRunsPolicy());
    }
}

队列满了用 CallerRunsPolicy（调用线程自己跑），避免丢报文；压测时观察队列深度。

9. 与 Spring 集成要点汇总

点	做法
Handler 生命周期	做成 `@Component` 单例 Bean（无状态），由多个 Channel 共享
连接维度状态	存 `Channel.attr(...)` 或独立 `ChannelManager`（Map<deviceId, Channel>）
业务依赖	在 Handler 里 `@Resource` 注入 service/线程池
Netty 启动/关闭	`@PostConstruct` 启动 + `@PreDestroy` 优雅关闭，跟着 Spring 生命周期走
配置项	端口、心跳时长、线程池大小放 `application.yml`，可进 Nacos 配置中心
项目启动方式	Spring Boot 主应用启动即拉起 Netty（无 HTTP 也正常）

10. InfluxDB3 写入（TelemetryWriteService）

@Service
public class TelemetryWriteService {
    private InfluxDBClient3 client;

    @PostConstruct
    public void init() {
        client = InfluxDBClient3.builder()
            .host("http://127.0.0.1:8181")
            .token("<读写token>")
            .database("iot_data")
            .build();
    }

    // 设备遥测：line protocol
    public void writeTelemetry(DeviceMessage m) {
        Map<String, Object> b = m.getBody();
        String lp = String.format(
            "telemetry,device_id=%s,type=%s,station_id=%s temperature=%s,voltage=%s,current=%s,soc=%s",
            m.getDeviceId(), b.get("type"), b.get("stationId"),
            b.get("temperature"), b.get("voltage"), b.get("current"), b.get("soc"));
        client.write(WriteRequest.builder().body(lp).build());
    }
}

性能：MVP 可用上面的同步写；正式压测建议批量攒批（条数阈值 + 时间阈值触发），并对写入失败做熔断 → 落 t_device_buffer → 恢复回补。

⚠️ 具体 API 以 influxdb3-client-java 官方文档为准，上述为概念示意。

11. 测试方法

单元测试解码：构造字节数组（20 字节头 + JSON 体），调用 ProtocolCodec.decode 断言解析结果。

端到端：启动 iot-gateway-service，用 device-simulator 连接 9000 端口。

# 观察 Netty 日志是否收到并解析
# 查 InfluxDB3 是否写入
influxdb3 query "SELECT * FROM telemetry" -d iot_data

粘包验证：模拟器高频连续发包，确认每条都正确解析。
离线验证：停掉某设备心跳，60 秒后状态变离线、收到 device.offline。
压测：模拟器 500 路连接、1000 报文/秒，观察业务线程池队列与 InfluxDB3 写入无堆积。

12. 容易踩的坑（实现时对照）

客户端依赖用错：用了 2.x 的 influxdb-client-java → 类名 API 全不对。
业务写在 EventLoop：阻塞导致其他设备掉线。
Handler 里存了可变状态：多连接共享 Bean，状态错乱。用 Channel.attr 或 ChannelManager。
报文头/长度字段解析错位：LengthFieldBasedFrameDecoder 的偏移要和报文头定义严格一致（长度字段偏移 16、4 字节）。
鉴权没移除：每个包都校验一次，浪费且无法接收非鉴权报文。
忘记 @PreDestroy 关闭：重启时端口被占用。

如果觉得文章对你有用，请随意赞赏

知识分享

Netty接入SpringBoot-iot-gateway实现

https://xiaochenblog.icu/archives/nettyjie-ru-springboot-iot-gatewayshi-xian

作者

xiaochen

发布于

2026-08-16 13:41

更新于

2026-08-16 13:35

许可协议

CC BY 4.0

Netty接入SpringBoot-iot-gateway实现

Netty 接入 SpringBoot 实现文档（iot-gateway-service）

1. 项目结构（iot-gateway-service）

2. pom.xml 依赖

3. 协议定义

3.1 报文结构（20 字节头 + JSON 体）

3.2 POJO

4. 解码器（ProtocolCodec）

5. Netty 服务端配置（NettyServerConfig）

6. 鉴权 Handler（AuthHandler）

7. 心跳 / 离线检测（HeartbeatHandler）

8. 业务分发 Handler（DeviceMsgHandler）

9. 与 Spring 集成要点汇总

10. InfluxDB3 写入（TelemetryWriteService）

11. 测试方法

12. 容易踩的坑（实现时对照）

作者

发布于

更新于

许可协议

评论