Netty编写TCP服务器和SpringBoot整合

- 日理万妓 2023-07-14 10:51 39阅读 0赞

`以前在写项目的时候用到过，这个tcp服务器的功能主要就是不间断的完成客户端发来的TCP连接请求，先是使用的阻塞式IO，然后又改为NIO，NIO写的时候出现了一些问题，就改用为Netty了`  
前言：此文仅提供思路，环境SpringBoot2.x，JDK8，Mysql5.7

### 导入Netty的依赖 ###

由于Netty并不是属于JDK自带的，它是一个开源的高性能的Java网络框架，是由JBoss提供的。  
注意：**如果就想使用Netty搭建服务端客户端就仅仅导入Netty就可，给出的是我项目部分的依赖**

<dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <scope>runtime</scope>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <optional>true</optional>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
            <dependency>
                <groupId>tk.mybatis</groupId>
                <artifactId>mapper-spring-boot-starter</artifactId>
                <version>2.1.5</version>
            </dependency>
    
            
            <dependency>
                <groupId>io.netty</groupId>
                <artifactId>netty-all</artifactId>
                <version>4.1.36.Final</version>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-configuration-processor</artifactId>
                <optional>true</optional>
            </dependency>
            
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.58</version>
            </dependency>

## Netty编写服务器端 ##

无论是客户端还是服务器端基本都分为三步走  
1.编写处理器：里面是服务器的处理逻辑，编写处理的时候需要注意接收的是那种类型的Msg  
2.编写一个初始化器：比如一些编码解码器，自定义处理器等等  
3. 编写服务端：服务端需要两个EventLoopGroup，一个是parent（boss），一个child（worker），还需要一个ServerBootStrap用来设定server启动相关，因为是非阻塞基于Channel的因此还需要一个ChannelFuture并绑定端口。  
在代码开始之前先简单的了解一下EventLoopGroup的执行原理，如下图可得每个EventLoopGroup都包含了一组的EventLoop，然后EventLoop是用来管理Channel的，Channel都要注册到（绑定到）EventLoop上。在我们平时用的阻塞式的IO是不能进行注册的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjA1OTY4_size_16_color_FFFFFF_t_70]  
那为什么需要两个EventLoopGroup来完成NIO呢？很好理解，一个EventLoopGroup无法同时进行大量的请求连接和响应工作。parent（boss）Group主要是处理连接请求，而child（worker）Group主要是处理响应。

具体代码如下（下面代码是我在项目开发前的测试版，具体逻辑可以自己在readChannel0中加）：  
MyServerHandler

package com.wrial.checkdev.netty;
    /*
     * @Author  Wrial
     * @Date Created in 0:11 2019/7/29
     * @Description  自定义消息截取处理器
     */
    
    import com.alibaba.fastjson.JSON;
    import com.alibaba.fastjson.JSONObject;
    import com.wrial.checkdev.service.LoginService;
    import com.wrial.checkdev.utils.SpringUtil;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    
    //这个泛型很重要，这取决于到底是那种方式的处理器
    public class MyServerHandler extends SimpleChannelInboundHandler<String> {
        /**
         * 
         * @param ctx 可以拿到本次处理的上下文信息
         * @param msg 就是获取到的信息
         * @throws Exception
         * 此处的逻辑就是将Netty和Spring整合后Netty并不在Spring的应用环境中
         * 需要编写一个SpringUtil来获取Spring内的Bean，用Json来规范字符串
         * 根据Type来判断请求类型，根据不同类型进行不同的数据库操作
         */
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
            System.out.println("客户端---" + ctx.channel().remoteAddress() + "------" + msg);
            LoginService loginService = (LoginService) SpringUtil.getBean("loginService");
            JSONObject jsonObject = JSON.parseObject(msg);
            String type = null;
            if (jsonObject.getString("type") != null) {
                type = jsonObject.getString("type");
            }
            switch (type) {
                case "1": {
                    String login = loginService.login(jsonObject.getString("telephone"), jsonObject.getString("password"));
                    ctx.writeAndFlush("from server code-------" + login);
                }
            }
            System.out.println("jsonObject====="+jsonObject);
    //        String login = loginService.login("admin", "admin");
        }
    
        /**
         * 
         * @param ctx 当前的应用上下文
         * @param cause Throwable是异常和Error的顶级接口,此处就是异常
         * @throws Exception
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            //有异常就关闭连接
            cause.printStackTrace();
            ctx.close();
        }
    
    }

MyServerInitializer  
主要是通过ch获取到PipeLine来添加处理器  
ChannelPipeline类是ChannelHandler实例对象的链表，用于处理或截获通道的接收和发送数据，每个新的通道Channel，都会创建一个新的ChannelPipeline，并将器pipeline附加到channel中。  
原理如下图：  
![在这里插入图片描述][20200314171825927.png]

package com.wrial.checkdev.netty;
    /*
     * @Author  Wrial
     * @Date Created in 0:06 2019/7/29
     * @Description   设置初始化器，主要是给pipeLine添加Handler
     */
    
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
    import io.netty.handler.codec.LengthFieldPrepender;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    import io.netty.util.CharsetUtil;
    
    public class MyServerInitializer  extends ChannelInitializer<SocketChannel> {
        // ch就可以对channel进行设置
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            //ChannelPipeline类是ChannelHandler实例对象的链表，用于处理或截获通道的接收和发送数据
            ChannelPipeline pipeline = ch.pipeline();
            // 也可以选择将处理器加到pipeLine的那个位置
            //一些限定和编码解码器
            pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
            pipeline.addLast(new LengthFieldPrepender(4));
            pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
            pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
            pipeline.addLast(new MyServerHandler());
        }
    }

NioServer  
其中要注意的是，handler和childHandler的区别：handler在初始化时就会执行，而childHandler会在客户端成功connect后才执行

package com.wrial.checkdev.netty;
    /*
     * @Author  Wrial
     * @Date Created in 23:59 2019/7/28
     * @Description Netty TCP Server
     */
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    
    
    public class MyServer {
    
    
        public void start() {
            System.out.println("netty Tcp 服务器启动");
    
            // 负责连接请求
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            // 负责事件响应
            EventLoopGroup workerGroup = new NioEventLoopGroup();
    
            try {
                // 服务器启动项
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                //handler是针对bossGroup，childHandler是针对workerHandler
                serverBootstrap.group(bossGroup, workerGroup)
                        // 选择nioChannel
                        .channel(NioServerSocketChannel.class)
                        // 日志处理 info级别
                        .handler(new LoggingHandler(LogLevel.INFO))
                        // 添加自定义的初始化器
                        .childHandler(new MyServerInitializer());
                // 端口绑定
                ChannelFuture channelFuture = serverBootstrap.bind(9999).sync();
                //该方法进行阻塞,等待服务端链路关闭之后继续执行。
                //这种模式一般都是使用Netty模块主动向服务端发送请求，然后最后结束才使用
                channelFuture.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                // 关闭
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
    
        }
    }

## Netty编写客户端 ##

客户端代码和服务端大致相似，下面就直接贴出来  
MyClientHandler

package com.wrial.checkdev.netty;
    /*
     * @Author  Wrial
     * @Date Created in 0:27 2019/7/29
     * @Description ClientHandler
     */
    
    import com.alibaba.fastjson.JSON;
    import com.alibaba.fastjson.JSONObject;
    import com.wrial.checkdev.entity.Admin;
    import com.wrial.checkdev.tcpdto.LoginDTO;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    
    import java.time.LocalDateTime;
    
    public class MyClientHandler extends SimpleChannelInboundHandler<String> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
            System.out.println("服务端"+ctx.channel().remoteAddress()+"-----"+msg);
            System.out.println("client receive"+msg);
        }
    
        //当连接建立好的使用调用
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            // ctx.channel().writeAndFlush("你好，服务端！");
            // 模拟用户注册
            LoginDTO loginDTO = new LoginDTO();
            loginDTO.setTelephone("admin");
            loginDTO.setPassword("admin");
            loginDTO.setType("1");
            String jsonString = JSON.toJSONString(loginDTO);
            ctx.writeAndFlush(jsonString);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

MyClientInitializer

package com.wrial.checkdev.netty;
    /*
     * @Author  Wrial
     * @Date Created in 0:24 2019/7/29
     * @Description ClientInitializer
     */
    
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
    import io.netty.handler.codec.LengthFieldPrepender;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    import io.netty.util.CharsetUtil;
    
    public class MyClientInitializer extends ChannelInitializer<SocketChannel> {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            //解码器
            pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
            //编码器
            pipeline.addLast(new LengthFieldPrepender(4));
            pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
            pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
            pipeline.addLast(new MyClientHandler());
        }
    }

MyClient

package com.wrial.checkdev.netty;
    /*
     * @Author  Wrial
     * @Date Created in 0:18 2019/7/29
     * @Description Client
     */
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioSocketChannel;
    
    public class MyClient {
    
        public static void main(String[] args) {
            EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
    
            try {
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(eventLoopGroup)
                        .channel(NioSocketChannel.class)
                        //也可以使用匿名类
                        .handler(new MyClientInitializer());
    
                ChannelFuture channelFuture = bootstrap.connect("localhost", 9999).sync();
                //要是closeFuture
                channelFuture.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                eventLoopGroup.shutdownGracefully();
            }
        }
    }

这个是在SpringBoot前提下进行Netty的整合，保证在SpringBoot项目运行基础上对Netty服务器也进行服务。

package com.wrial.checkdev;
    
    import com.wrial.checkdev.netty.MyServer;
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import tk.mybatis.spring.annotation.MapperScan;
    
    @SpringBootApplication
    @MapperScan(basePackages = "com.wrial.checkdev.mapper")
    public class CheckDevApplication {
    
        public static void main(String[] args)   {
            SpringApplication.run(CheckDevApplication.class, args);
            MyServer myServer = new MyServer();
            myServer.start();
        }
    
    }

至于SpringUtils的编写和分析可以在Spring源码专题中看，使用到了ApplicationContextAware接口，动态的获取application，此处不多说了。  
测试阶段：可以看到Netty服务器和Tomcat服务器共存了  
启动服务器：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjA1OTY4_size_16_color_FFFFFF_t_70 1]  
启动客户端模拟用户注册  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjA1OTY4_size_16_color_FFFFFF_t_70 2]  
客户端内容：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjA1OTY4_size_16_color_FFFFFF_t_70 3]  
服务端内容：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjA1OTY4_size_16_color_FFFFFF_t_70 4]  
这样就满足了项目需求了，真实的项目并不是这样，这个只是最初的测试版（感觉我这个想法很奇妙哈哈）。  
这就是今天要分享的内容。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjA1OTY4_size_16_color_FFFFFF_t_70]: /images/20230529/9eda97a8ee984c66b2016bf90e218f90.png
[20200314171825927.png]: /images/20230529/883fcbba7db04365b6fccfb4960755ee.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjA1OTY4_size_16_color_FFFFFF_t_70 1]: /images/20230529/023c5b96dcd349dfbf2e1c2ff5c6e114.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjA1OTY4_size_16_color_FFFFFF_t_70 2]: /images/20230529/e74a090608394eef9ab2f3d0e7473a2f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjA1OTY4_size_16_color_FFFFFF_t_70 3]: /images/20230529/e92dc3ca45b64faebf66f516d903ec3d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQyNjA1OTY4_size_16_color_FFFFFF_t_70 4]: /images/20230529/e4c238fc19c942ed88f6f706c00ec395.png