���迎来到我的博客，代码的世界里，每一行都是一个故事

Spring Boot与Netty打造TCP服务端

• • 前言
  • 功能目标
  • 项目实现
  • • maven坐标
    • 构建自定义Handler
    • ChannelInitializer实现
    • server实现

      前言

      在物联网时代，设备之间的通信变得愈发重要。本文将带你踏上一场关于如何用Spring Boot和Netty搭建TCP服务端的冒险之旅。无论是智能家居、工业自动化还是其他物联网应用，构建一个稳健的通信桥梁将成为连接未来的关键。

      功能目标

      • 实现springboot+netty整合TCP服务端（基础）
      • 实现消息回复功能
      • 实现消息太长导致的粘包问题(比如发送一个base64的图片信息)
      • 实现在自定义Handler中注入spring的bean
      • 保证完成任务，哈哈哈哈哈

        项目实现

        maven坐标

          io.netty  netty-common  4.1.79.Final

        构建自定义Handler

        import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;import io.netty.util.CharsetUtil;import lombok.extern.slf4j.Slf4j;import java.io.IOException;import java.net.InetSocketAddress;/** * I/O数据读写处理类 * * @author xiaobo */@Slf4jpublic class CarTcpNettyChannelInboundHandlerAdapter extends ChannelInboundHandlerAdapter {    /**     * 从客户端收到新的数据时，这个方法会在收到消息时被调用     *     * @param ctx     * @param msg     */    @Override    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception, IOException {        // 这里是在前面的DelimiterBasedFrameDecoder转为了ByteBuf，验证是否是ByteBuf        if (msg instanceof ByteBuf) {            ByteBuf byteBuf = (ByteBuf) msg;            try {                String receivedData = byteBuf.toString(CharsetUtil.UTF_8);                // 接收完整数据                handleReceivedData(receivedData);            } finally {                // 释放 ByteBuf 占用的资源                byteBuf.release();                // 回复消息                ctx.writeAndFlush(Unpooled.copiedBuffer("收到over", CharsetUtil.UTF_8));            }        }    }    private void handleReceivedData(String receivedData) {        // 数据处理        // 这里如果想实现spring中bean的注入，可以用geBean的方式获取    }    /**     * 从客户端收到新的数据、读取完成时调用     *     * @param ctx     */    @Override    public void channelReadComplete(ChannelHandlerContext ctx) throws IOException {        log.info("channelReadComplete");        ctx.flush();    }    /**     * 当出现 Throwable 对象才会被调用，即当 Netty 由于 IO 错误或者处理器在处理事件时抛出的异常时     *     * @param ctx     * @param cause     */    @Override    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws IOException {        cause.printStackTrace();        ctx.close();// 抛出异常，断开与客户端的连接    }    /**     * 客户端与服务端第一次建立连接时 执行     *     * @param ctx     * @throws Exception     */    @Override    public void channelActive(ChannelHandlerContext ctx) throws Exception, IOException {        super.channelActive(ctx);        ctx.channel().read();        InetSocketAddress socket = (InetSocketAddress) ctx.channel().remoteAddress();        String clientIp = socket.getAddress().getHostAddress();        // 此处不能使用ctx.close()，否则客户端始终无法与服务端建立连接        System.out.println("channelActive:" + clientIp + ctx.name());      // 这里是向客户端发送回应        ctx.writeAndFlush(Unpooled.copiedBuffer("收到over", CharsetUtil.UTF_8));    }    /**     * 客户端与服务端 断连时 执行     *     * @param ctx     * @throws Exception     */    @Override    public void channelInactive(ChannelHandlerContext ctx) throws Exception, IOException {        super.channelInactive(ctx);        InetSocketAddress socket = (InetSocketAddress) ctx.channel().remoteAddress();        String clientIp = socket.getAddress().getHostAddress();        // 断开连接时，必须关闭，否则造成资源浪费，并发量很大情况下可能造成宕机        ctx.close();        log.info("channelInactive:{}", clientIp);    }    /**     * 服务端当read超时, 会调用这个方法     *     * @param ctx     * @param evt     * @throws Exception     */    @Override    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception, IOException {        super.userEventTriggered(ctx, evt);        InetSocketAddress socket = (InetSocketAddress) ctx.channel().remoteAddress();        String clientIp = socket.getAddress().getHostAddress();        ctx.close();// 超时时断开连接        log.info("userEventTriggered:" + clientIp);    }    @Override    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {        log.info("channelRegistered");    }    @Override    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {        log.info("channelUnregistered");    }    @Override    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {        log.info("channelWritabilityChanged");    }}

        ChannelInitializer实现

        import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;import io.netty.channel.Channel;import io.netty.channel.ChannelInitializer;import io.netty.handler.codec.DelimiterBasedFrameDecoder;/** * description: 
        通道初始化
         * * @author bo * @version 1.0 * @date 2024/2/27 16:13 */public class CarTcpNettyChannelInitializer extends ChannelInitializer {    @Override    protected void initChannel(Channel ch) throws Exception {        ByteBuf delemiter = Unpooled.buffer();        delemiter.writeBytes("$".getBytes());        // 这里就是解决数据过长问题，而且数据是以$结尾的        ch.pipeline().addLast(new DelimiterBasedFrameDecoder(907200, true, true, delemiter));        // 自定义ChannelInboundHandlerAdapter        ch.pipeline().addLast(new CarTcpNettyChannelInboundHandlerAdapter());    }}

        server实现

        import io.netty.bootstrap.ServerBootstrap;import io.netty.channel.AdaptiveRecvByteBufAllocator;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelOption;import io.netty.channel.EventLoopGroup;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioServerSocketChannel;import lombok.extern.slf4j.Slf4j;/** * description: 
        netty创建的TCP
         * * @author bo * @version 1.0 * @date 2024/2/27 16:25 */@Slf4jpublic class CarTcpNettyServer {    public void bind(int port) throws Exception {        // 配置服务端的NIO线程组        // NioEventLoopGroup 是用来处理I/O操作的Reactor线程组        // bossGroup：用来接收进来的连接，workerGroup：用来处理已经被接收的连接,进行socketChannel的网络读写，        // bossGroup接收到连接后就会把连接信息注册到workerGroup        // workerGroup的EventLoopGroup默认的线程数是CPU核数的二倍        EventLoopGroup bossGroup = new NioEventLoopGroup(1);        EventLoopGroup workerGroup = new NioEventLoopGroup();        try {            ServerBootstrap serverBootstrap = new ServerBootstrap();            serverBootstrap.group(bossGroup, workerGroup)                    .channel(NioServerSocketChannel.class)                    // netty 默认数据包传输大小为1024字节, 设置它可以自动调整下一次缓冲区建立时分配的空间大小，避免内存的浪费    最小  初始化  最大 (根据生产环境实际情况来定)                    // 使用对象池，重用缓冲区                    .option(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(64, 10496, 1048576))                    .childOption(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(64, 10496, 1048576))                    // 设置 I/O处理类,主要用于网络I/O事件，记录日志，编码、解码消息                    .childHandler(new CarTcpNettyChannelInitializer());            log.info("");            // 绑定端口，同步等待成功            ChannelFuture f = serverBootstrap.bind(port).sync();            // 等待服务器监听端口关闭            f.channel().closeFuture().sync();        } finally {            // 退出，释放线程池资源            bossGroup.shutdownGracefully();            workerGroup.shutdownGracefully();        }    }}