基于Netty实现一个服务端/客户端通信demo

雨点打透心脏的1/2处 2022-10-20 01:18 205阅读 0赞

> 原创不易，转载请注明出处

### 文章目录 ###

*   *   *  前言
         *  demo程序编写
         *   *  maven的pom依赖
             *  服务端程序
             *  客户端程序
             *  测试
         *  总结

--------------------

### 前言 ###

本文主要是使用netty这个高性能网络通信框架写一个服务端、客户端通信的demo，体验下基于netty的网络编程是什么样子的，在此之前需要有java nio基础，毕竟netty就是封装的java nio，写完之后介绍一下netty核心组件底层运行原理。

### demo程序编写 ###

#### maven的pom依赖 ####

<dependency>
        <groupId>io.netty</groupId>
        <artifactId>netty-all</artifactId>
        <version>4.1.6.Final</version>
    </dependency>

加到你pom文件中可以

#### 服务端程序 ####

public class NettyServer { 
        public static void main(String[] args) { 
            EventLoopGroup  bossGroup=new NioEventLoopGroup(1);
            EventLoopGroup  workerGroup=new NioEventLoopGroup();
            try { 
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap
                        .group(bossGroup,workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG,1024)
                        .option(ChannelOption.SO_REUSEADDR,true)
                        .childOption(ChannelOption.TCP_NODELAY,true)
                        .childOption(ChannelOption.SO_RCVBUF,64*1024)
                        .childOption(ChannelOption.SO_SNDBUF,64*1024)
                        .childHandler(new ChannelInitializer() { 
                            @Override
                            protected void initChannel(Channel channel) throws Exception { 
                                ChannelPipeline pipeline = channel.pipeline();
                                pipeline.addLast(new NettyServerHandler());
                            }
                        });
                ChannelFuture channelFuture= serverBootstrap.bind(10009).sync();
                channelFuture.channel().closeFuture().sync();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }finally { 
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
    
        }
    }

稍微解释一下  
首先是创建了2个EventLoopGroup，这个玩意其实你可以看做是线程池，线程组都可以，从名字上看，专门用作循环事件的，我这边给起了2个名字，分别是bossGroup与workerGroup（你也可以管他们叫parentGroup 与childGroup），这个两个名字是有实际含义的，表示任务分工不一样，如果你使用过java nio写过程序，或者是了解reactor网络模型的话，bossGroup其实就是专门用来创建连接的，而workerGroup专门就是接收请求，处理请求，发送响应的，也就是专门干活的，如果你对reactor网络模型不了解的话，建议网上搜下看下，当然也可以看下我写的《[深度解析kafka broker网络模型运行原理][kafka broker]》kafka网络模型，尤其是那张图。看完之后印象深刻，bossGroup里面的线程其实就是acceptor线程，workerGroup线程就是processor线程，专门处理接收请求，返回响应的。

ServerBootstrap这个可以理解为服务器，它主要就是创建服务器使用的，构建服务器程序的，这里将这两个group传入进去，channel的话就是使用哪种类型的channel，他底层会创建这个channel的对象，这里这个NioServerSocketChannel 其实底层就是封装了java nio 的ServerSocketChannel，option的话就是服务端程序的设置，childOption 是与客户端连接的一些设置，childHandler其实就是与某个客户端通信的时候，比如读取客户端发送过来的数据，读到这些数据之后，进行的一些处理，这里是一个初始化的handler，也就是当与某个客户端建立连接完事的时候，将这个连接channel丢给某个workerGroup中的某个线程进行注册，并初始化处理链，这里面这个pipeline 就是一条处理链，每个连接有一个pipeline，我们这里往pipeline处理链后面添加了一个处理器，也就是NettyServerHandler ，我们下面看下。  
最后就是bind了，绑定端口，监听端口，这个bind底层做了很多事情，先是创建NioServerSocketChannel，其实就是创建一个ServerSocketChannel，接着就是将ServerSocketChannel注册到Selector上面，接着就是bind端口，这个bossGroup线程就会不停的轮询看看有没有accept事件的发生。

如果有客户端连接的话，就会select到，然后建立连接，将连接交给workerGroup的某个线程进行注册到对应selector 上面去，然后也是不停的轮询selector，看看有没有读写事件的发生。  
需要注意的是 每个EventLoop线程都有自己对应的selector，并不是共用的一个。

public class NettyServerHandler extends ChannelInboundHandlerAdapter { 
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
            ByteBuf byteBuf= (ByteBuf)msg;
            byte[] body=new byte[byteBuf.readableBytes()];
            byteBuf.readBytes(body);
            System.out.println(new String(body));
            byte[] responseBytes = "hi,客户端,我是服务端".getBytes();
            ctx.channel().writeAndFlush(Unpooled.wrappedBuffer(responseBytes));
        }
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            cause.printStackTrace();
        }
    }

这个handler是放到pipeline里面的，也就是某个连接有读写事件的时候，worker线程就会调用pipeline处理链进行处理，这里重写了channelRead方法，就是某个客户端发送过来数据，然后worker线程拿到这个read事件的发生，就会调用pipeline进行数据的处理，这里就是读出来数据，然后打印，最后又发送响应给客户端。

#### 客户端程序 ####

public class NettyClient { 
    
        public static void main(String[] args) { 
    
    
            EventLoopGroup  bossGroup=new NioEventLoopGroup();
            try { 
                Bootstrap bootstrap = new Bootstrap();
                bootstrap
                        .group(bossGroup)
                        .channel(NioSocketChannel.class)
                        .option(ChannelOption.SO_RCVBUF, 64 * 1024)
                        .option(ChannelOption.SO_SNDBUF, 64 * 1024)
                        .option(ChannelOption.SO_REUSEADDR, true)
                        .option(ChannelOption.TCP_NODELAY, true)
                        .handler(new ChannelInitializer<Channel>() { 
                            @Override
                            protected void initChannel(Channel channel) throws Exception { 
                                ChannelPipeline pipeline = channel.pipeline();
                                pipeline.addLast(new NettyClientHandler());
                            }
                        });
                ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 10009).sync();
                channelFuture.channel().closeFuture().sync();
            }catch (Exception e){ 
                e.printStackTrace();
            }finally { 
                bossGroup.shutdownGracefully();
            }
    
        }
    }

客户端程序跟服务端程序差不多，就是EventLoopGroup组少了一个，然后对应的channel就变成了NioSocketChannel  
最后是connect进行连接。这里需要注意的是，我们将handler放到了bossGroup的pipeline处理链中，这个bossGroup干的事情就是建立连接，发送数据，读取响应。

public class NettyClientHandler extends ChannelInboundHandlerAdapter { 
    
    
        /** * 连接建立的完成 * @param ctx * @throws Exception */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception { 
            byte[] bytes = "hi,服务端，我是客户端！".getBytes();
            ByteBuf byteBuf = Unpooled.wrappedBuffer(bytes);
            ctx.channel().writeAndFlush(byteBuf);
        }
    
        /** * 读取数据 * @param ctx * @param msg * @throws Exception */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
            ByteBuf responseByteBuf= (ByteBuf)msg;
            byte[] responseByte=new byte[responseByteBuf.readableBytes()];
            responseByteBuf.readBytes(responseByte);
            System.out.println(new String(responseByte));
        }
    
    
        /** * 异常 * @param ctx * @param cause * @throws Exception */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            cause.printStackTrace();
        }
    }

channelActive 这个方法是当连接建立完成的时候，然后向服务端发送了hi,服务端，我是客户端！ 这句话，当收到服务端响应的时候，就会走channelRead方法，这里就是读取到数据，然后打印一下。

#### 测试 ####

先启动服务器程序，然后再启动客户端程序。  
服务器端：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]  
客户端：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]

### 总结 ###

本文主要是使用netty实现了一个客户端服务端通信的小demo，然后详细介绍了一下服务端程序的运行原理。

[kafka broker]: https://inetyoung.blog.csdn.net/article/details/112307717
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]: /images/20221004/95c27a261ade443184d8f7b85ecddffd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70 1]: /images/20221004/2d95deacdf9b41668a9a4752aec46235.png