<!-- Netty核心依赖：提供NIO网络通信能力 -->
<dependency>
  <groupId>io.netty</groupId>
  <artifactId>netty-all</artifactId>
  <!-- 选择4.1.x最新稳定版，保证功能完整性和兼容性 -->
  <version>4.1.100.Final</version>
</dependency>
<!-- 监控组件：用于收集网络流量指标 -->
<dependency>
  <groupId>io.micrometer</groupId>
  <artifactId>micrometer-core</artifactId>
  <!-- 版本需与Spring Boot版本匹配 -->
  <version>1.11.5</version>
  <!-- optional表示不会被传递依赖 -->
  <optional>true</optional>
</dependency>

1.2. YAML配置文件

tcp:
  server:
    port: 8888       # 监听端口（必须>1024）
    boss-threads: 1   # 固定1线程处理连接请求
    worker-threads: 0 # 0=自动计算（CPU核数*2）
  client:
    host: 127.0.0.1  # 服务端地址
    port: 8888        # 服务端端口
    reconnect-interval: 5    # 基础重连间隔（秒）
    max-reconnect-attempts: 10 # 0=无限重试

2. TCP服务端配置类

@Slf4j
@Component
@ConfigurationProperties(prefix = "tcp.server")
// 实现DisposableBean接口保证优雅关闭
public class TCPServer implements DisposableBean {
    /**
     *服务器监听端口
     */
    @Value("${tcp.server.port}")
    private int port;
    /**
     *Boss线程数（处理连接请求的线程数，通常设置为1）
     */
    @Value("${tcp.server.boss-threads}")
    private int bossThreads;
    /**
     *Worker线程数（处理IO操作的线程数，默认0表示CPU核数*2）
     */
    @Value("${tcp.server.worker-threads}")
    private int workerThreads;
    /**
     *Netty线程组（BOSS用于接受新连接）
     */
    private EventLoopGroup bossGroup;
    /**
     *Worker线程组用于处理已建立的连接
     */
    private EventLoopGroup workerGroup;
    /**
     *服务器通道实例
     */
    private Channel serverChannel;
    // 初始化方法：在Bean属性设置完成后执行
    @PostConstruct
    public void start() throws InterruptedException {
        // 初始化线程组NIO模式
        bossGroup = new NioEventLoopGroup(bossThreads);
        workerGroup = workerThreads > 0 ? 
        new NioEventLoopGroup(workerThreads) : 
        new NioEventLoopGroup();
        // 服务器启动引导类
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)// 指定NIO服务器套接字通道
        .childHandler(new ChannelInitializer<SocketChannel>() {// 初始化子通道处理器
            @Override
            protected void initChannel(SocketChannel ch) {
                // 构建处理管道
                buildPipeline(ch.pipeline());
            }
        }).option(ChannelOption.SO_BACKLOG, 1024)// TCP参数设置：等待连接队列大小（默认50）
        .childOption(ChannelOption.TCP_NODELAY, true)// 禁用Nagle算法（降低延迟）
        .childOption(ChannelOption.SO_KEEPALIVE, true);// 启用TCP keep-alive机制（检测连接存活）
        // 绑定端口并同步等待启动完成
        serverChannel = bootstrap.bind(port).sync().channel();
        log.info("TCP Server started on port: {}", port);
    }
    // 构建处理管道（关键：顺序决定处理流程）
    private void buildPipeline(ChannelPipeline pipeline) {
        /* 粘包/拆包解决方案：
         * LengthFieldBasedFrameDecoder参数说明：
         * maxFrameLength: 最大帧长度（字节）
         * lengthFieldOffset: 长度字段偏移量
         * lengthFieldLength: 长度字段字节数（4=int）
         * lengthAdjustment: 长度调整值
         * initialBytesToStrip: 跳过前导字节数
         */
        pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(
            1024, 0, 4, 0, 4));
        // 长度编码器（自动添加4字节长度头）
        pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
        /* 心跳检测机制：
         * readerIdleTimeSeconds: 读超时（秒）
         * writerIdleTimeSeconds: 写超时
         * allIdleTimeSeconds: 读写超时
         */
        pipeline.addLast("idleHandler", new IdleStateHandler(
            30, 0, 0, TimeUnit.SECONDS));
        // 字符串编码器（指定UTF-8字符集）
        pipeline.addLast("encoder", new StringEncoder(StandardCharsets.UTF_8));
        // 字符串解码器（自动处理ByteBuf到String的转换）
        pipeline.addLast("decoder", new StringDecoder(StandardCharsets.UTF_8));
        
        // 自定义业务处理器（必须放在最后）
        pipeline.addLast("handler", new TCPServerHandler());
    }
    // Spring Bean销毁时的清理操作
    @Override
    public void destroy() throws Exception {
        log.info("Shutting down TCP Server...");
        // 关闭服务器通道（同步等待）
        if (serverChannel != null) {
            serverChannel.close().sync();
        }
        // 优雅关闭线程组（先worker后boss）
        if (workerGroup != null) {
            workerGroup.shutdownGracefully().sync();
        }
        if (bossGroup != null) {
            bossGroup.shutdownGracefully().sync();
        }
        log.info("TCP Server resources released");
    }
}

3. TCP服务端处理器

@Slf4j
public class TCPServerHandler extends SimpleChannelInboundHandler<String> {// 继承SimpleChannelInboundHandler简化字符串处理
    
    // 线程安全的连接管理Map（Key为ChannelID）
    private static final ConcurrentMap<ChannelId, Channel> activeChannels = new ConcurrentHashMap<>();
    // 新连接建立时触发
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        Channel channel = ctx.channel();
        // 记录新连接
        activeChannels.put(channel.id(), channel);
        log.info("Client connected from: {}", channel.remoteAddress());
        
        // 发送欢迎消息（自动释放ByteBuf）
        ctx.writeAndFlush("Welcome to Server!\n");
    }
    // 接收到消息时触发（已自动解码为String）
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        try {
            log.debug("Received message: [{}]", msg);
            // 业务处理示例（应避免阻塞操作）
            String response = processMessage(msg);
            // 发送响应（自动编码为ByteBuf）
            ctx.writeAndFlush(response + "\n");
        } catch (Exception e) {
            log.error("Message processing failed", e);
            // 发送错误响应
            ctx.writeAndFlush("ERROR: " + e.getMessage() + "\n");
        }
    }
    // 用户自定义事件触发（如心跳超时）
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent e = (IdleStateEvent) evt;
            // 处理读空闲超时
            if (e.state() == IdleState.READER_IDLE) {
                log.warn("Client {} timeout", ctx.channel().remoteAddress());
                // 关闭超时连接
                ctx.close();
            }
        }
    }
    // 异常处理
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // 分类处理异常类型
        if (cause instanceof DecoderException) {
            log.warn("Decoding error: {}", cause.getMessage());
        } else if (cause instanceof IOException) {
            log.info("Client force closed connection");
        } else {
            log.error("Unexpected error", cause);
        }
        // 关闭问题连接
        ctx.close();
    }
    // 连接断开时触发
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        // 移除连接记录
        activeChannels.remove(ctx.channel().id());
        log.info("Client disconnected: {}", ctx.channel().remoteAddress());
    }
    // 示例消息处理逻辑
    private String processMessage(String msg) {
        // 实现业务逻辑（建议异步处理耗时操作）
        return "Processed: " + msg.toUpperCase();
    }
}

4. 粘包/拆包

在TCP通信中，粘包(TCP Stickiness)和拆包(Unpacking) 是必须解决的核心问题，因为TCP是面向流的协议，数据在传输过程中会被分割成多个包，或者多个小的数据包被合并成一个大的包发送。例如：

当发送方连续发送多个小数据包时，接收方可能一次性接收到多个包，这就是粘包。
一个大数据包被拆分成多个小包接收，这就是拆包。

下面从技术原理角度详细解释：

4.1. 粘包/拆包产生原理

TCP协议特性：

流式传输：TCP协议没有消息边界概念，数据以字节流形式传输
缓冲区机制：发送端Nagle算法会合并小数据包，接收端缓冲区可能累积多个包
网络MTU限制：数据包超过最大传输单元（MTU）会被自动分片

典型场景：假设连续发送3个数据包（P1, P2, P3），接收端可能出现：

粘包：收到 [P1+P2][P3] 或 [P1][P2+P3]
拆包：收到 [P1_part1][P1_part2+P2]
混合情况：[P1_part1][P1_part2+P2][P3]

4.2. 解决方案

4.2.1. 传统方案（不推荐）

方案	原理	缺点
固定长度	所有数据包长度相同	浪费带宽，不灵活
分隔符（如`\n`）	特殊字符标识消息结束	需要转义字符，处理复杂
固定头声明包长度	头部字段声明数据部分长度	需要自定义编解码

4.2.2. Netty的优化方案

通过LengthFieldBasedFrameDecoder + LengthFieldPrepender组合，实现：

自动拆包：基于长度字段精准切分数据流
零拷贝处理：基于ByteBuf的指针操作，无内存复制
协议无关性：适用于各种二进制/文本协议

4.3. 代码参数深度解析

// 解码器
new LengthFieldBasedFrameDecoder(
    1024,    // maxFrameLength: 最大允许的帧长度（超过抛异常）
    0,       // lengthFieldOffset: 长度字段偏移量（从第0字节开始）
    4,       // lengthFieldLength: 长度字段占4字节（int类型）
    0,       // lengthAdjustment: 长度调整值=0（长度字段值=数据长度）
    4        // initialBytesToStrip: 跳过前4字节（丢弃长度字段）
)
// 编码器
new LengthFieldPrepender(4) // 添加4字节长度头

4.4. 工作流程

假设发送消息内容为 "HELLO"（5字节）：

发送端处理：

LengthFieldPrepender 计算长度：5 → 0x00000005
添加4字节头 → 最终发送：[0x00 0x00 0x00 0x05][H][E][L][L][O]

接收端处理：

LengthFieldBasedFrameDecoder 读取前4字节 → 0x00000005
截取后续5字节 → 得到完整消息 "HELLO"
丢弃长度字段（initialBytesToStrip=4） → 传递给后续Handler的是净荷数据

4.5. 关键技术点

ByteBuf的指针操作

// 在LengthFieldBasedFrameDecoder源码中：
int actualLength = length + lengthFieldEndOffset;
ByteBuf frame = extractFrame(ctx, in, index, actualLength);
in.readerIndex(index + actualLength); // 移动读指针

通过直接操作ByteBuf的读写指针，避免内存复制

内存保护机制

maxFrameLength 防止恶意超大包导致OOM
超过阈值会触发 TooLongFrameException

长度调整公式

实际截取长度 = 长度字段值 + lengthFieldOffset + lengthFieldLength + lengthAdjustment

通过调整参数可处理复杂协议头（如包含版本号等额外字段）

4.6. 与其他方案的对比

特性	长度字段法	分隔符法
处理效率	⭐⭐⭐⭐⭐（O(1)复杂度）	⭐⭐（需遍历查找）
协议灵活性	支持任意二进制协议	仅适用文本协议
内存消耗	固定头部开销	需处理转义字符
复杂协议支持	通过参数调整轻松支持	难以扩展

4.7. 生产环境注意事项

长度字段大小端问题

需与协议设计方约定字节序（推荐使用网络字节序：Big-Endian）

超长包防御

// 建议设置合理的maxFrameLength
new LengthFieldBasedFrameDecoder(1024 * 1024, ...) // 1MB上限

版本兼容性

在协议头预留version字段，应对未来协议升级

// 示例：lengthFieldOffset=2（跳过2字节版本号）
new LengthFieldBasedFrameDecoder(1024, 2, 4, ...)

通过这种基于长度字段的解决方案，可以完美解决TCP粘包/拆包问题，同时保持协议的高效性和扩展性，是工业级网络通信的首选方案。

5. TCP客户端配置类

@Slf4j
@Component
@ConfigurationProperties(prefix = "tcp.client")
public class TCPClient implements DisposableBean {
    // 原子化连接状态（保证线程可见性）
    private final AtomicBoolean connected = new AtomicBoolean(false);
    
    // 定时重连线程池
    private ScheduledExecutorService reconnectExecutor;
    
    // Netty IO线程组
    private EventLoopGroup eventLoopGroup;
    
    // 客户端通道实例
    private Channel clientChannel;
    /**
     *客户端地址
     */
    @Value("${tcp.client.host}")
    private String host;
    /**
     *客户端端口
     */
    @Value("${tcp.client.port}")
    private int port;
    /**
     *基础重连间隔（秒）
     */
    @Value("${tcp.client.reconnect-interval}")
    private int reconnectInterval;
    /**
     *最大重试次数
     */
    @Value("${tcp.client.max-reconnect-attempts}")
    private int maxReconnectAttempts;
    // 初始化方法（在属性注入后执行）
    @PostConstruct
    public void init() {
        // 创建单线程定时任务执行器
        reconnectExecutor = Executors.newSingleThreadScheduledExecutor(
            r -> new Thread(r, "tcp-reconnect")
        );
        startConnection();
    }
    // 初始化连接
    private void startConnection() {
        // 创建NIO线程组（2个线程处理IO）
        eventLoopGroup = new NioEventLoopGroup(2);
        
        Bootstrap bootstrap = new Bootstrap()
                .group(eventLoopGroup)
                .channel(NioSocketChannel.class) // 客户端使用NIO套接字
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        buildPipeline(ch.pipeline());
                    }
                });
        // 启动带重试机制的连接
        connectWithRetry(bootstrap, 0);
    }
    // 带指数退避的重连机制
    private void connectWithRetry(Bootstrap bootstrap, int attempt) {
        bootstrap.connect(host, port).addListener((ChannelFuture future) -> {
            if (future.isSuccess()) {
                log.info("Connected to {}:{}", host, port);
                connected.set(true);
                clientChannel = future.channel();
            } else if (attempt < maxReconnectAttempts) {
                // 计算退避时间（指数增长）
                int delay = reconnectInterval * (1 << attempt);
                log.warn("Connection failed, retrying in {}s (attempt {})", delay, attempt+1);
                // 调度重试任务
                reconnectExecutor.schedule(
                    () -> connectWithRetry(bootstrap, attempt + 1),
                    delay, TimeUnit.SECONDS
                );
            } else {
                log.error("Max reconnect attempts({}) reached", maxReconnectAttempts);
                connected.set(false);
            }
        });
    }
    // 构建客户端处理管道
    private void buildPipeline(ChannelPipeline pipeline) {
        // 与服务器相同的编解码器配置
        pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(1024, 0, 4, 0, 4));
        pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
        
        // 客户端写空闲检测（15秒发送心跳）
        pipeline.addLast("idleHandler", new IdleStateHandler(0, 15, 0));
        
        pipeline.addLast("encoder", new StringEncoder(StandardCharsets.UTF_8));
        pipeline.addLast("decoder", new StringDecoder(StandardCharsets.UTF_8));
        
        // 客户端业务处理器
        pipeline.addLast("handler", new TCPClientHandler());
    }
    // 发送消息（非阻塞）
    public void sendMessage(String message) {
        if (connected.get() && clientChannel != null) {
            // 添加发送结果监听器
            clientChannel.writeAndFlush(message).addListener(future -> {
                if (!future.isSuccess()) {
                    log.error("Message send failed", future.cause());
                    // 可在此处实现消息重试队列
                }
            });
            log.debug("Sent message: {}", message);
        } else {
            log.warn("Message dropped: No active connection");
            // 可扩展消息缓存队列
        }
    }
    // 资源清理
    @Override
    public void destroy() {
        log.info("Shutting down TCP Client...");
        if (clientChannel != null) {
            clientChannel.close();
        }
        if (eventLoopGroup != null) {
            eventLoopGroup.shutdownGracefully();
        }
        if (reconnectExecutor != null) {
            reconnectExecutor.shutdownNow();
        }
    }
}

6. TCP客户端处理器

@Slf4j
public class TCPClientHandler extends SimpleChannelInboundHandler<String> {
    // 接收到服务器响应
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        log.info("Server response: {}", msg);
        // 实现业务逻辑（如更新状态、触发回调等）
    }
    // 处理自定义事件（心跳）
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent e = (IdleStateEvent) evt;
            if (e.state() == IdleState.WRITER_IDLE) {
                log.debug("Sending heartbeat...");
                ctx.writeAndFlush("HEARTBEAT\n");
            }
        }
    }
    // 异常处理
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("Connection error: {}", cause.getMessage());
        ctx.close();
    }
    // 连接断开时触发
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        log.info("Connection closed");
        // 可在此触发重连逻辑
    }
}

7. 连接测试

编辑

SpringBoot × TCP 极速开发指南：工业级TCP通信协议栈操作手册

SpringBoot Integrate TCP

1. 项目配置与依赖管理

1.1. Maven依赖管理

1.2. YAML配置文件

2. TCP服务端配置类

3. TCP服务端处理器

4. 粘包/拆包

4.1. 粘包/拆包产生原理

4.2. 解决方案

4.2.1. 传统方案（不推荐）

4.2.2. Netty的优化方案

4.3. 代码参数深度解析

4.4. 工作流程

4.5. 关键技术点

4.6. 与其他方案的对比

4.7. 生产环境注意事项

5. TCP客户端配置类

6. TCP客户端处理器

7. 连接测试

热门文章

最新文章

相关课程

相关电子书

相关实验场景

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

SpringBoot × TCP 极速开发指南：工业级TCP通信协议栈操作手册

SpringBoot Integrate TCP

1. 项目配置与依赖管理

1.1. Maven依赖管理

1.2. YAML配置文件

2. TCP服务端配置类

3. TCP服务端处理器

4. 粘包/拆包

4.1. 粘包/拆包产生原理

4.2. 解决方案

4.2.1. 传统方案（不推荐）

4.2.2. Netty的优化方案

4.3. 代码参数深度解析

4.4. 工作流程

4.5. 关键技术点

4.6. 与其他方案的对比

4.7. 生产环境注意事项

5. TCP客户端配置类

6. TCP客户端处理器

7. 连接测试

热门文章

最新文章

相关课程

相关电子书

相关实验场景