DangerWind-RPC-framework---八、Netty网络通信

07-21 1256阅读

客户端与服务端之间发送数据需要进行网络通信，而Netty恰好是一款出色的网络通信框架，Netty底层实现依赖于NIO，其工作线程组EventLoopGroup中的每一个EventLoop可以理解为一个线程，与NIO类似，EventLoopGroup的工作也是基于事件响应的，可以理解为bossEventLoop负责监听连接事件，并建立连接，众多workEventLoop监听读事件，在channel可读时读入数据并进行处理。

在写出与读入数据的过程中，数据会经过流水线上的一道道Handler来处理，首先以客户端为例：

   public NettyRpcClient() {
        // initialize resources such as EventLoopGroup, Bootstrap
        eventLoopGroup = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new LoggingHandler(LogLevel.INFO))
                //  The timeout period of the connection.
                //  If this time is exceeded or the connection cannot be established, the connection fails.
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .handler(new ChannelInitializer() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ChannelPipeline p = ch.pipeline();
                        // If no data is sent to the server within 15 seconds, a heartbeat request is sent
                        p.addLast(new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS));
                        p.addLast(new RpcMessageEncoder());
                        p.addLast(new RpcMessageDecoder());
                        p.addLast(new NettyRpcClientHandler());
                    }
                });
        this.serviceDiscovery = ExtensionLoader.getExtensionLoader(ServiceDiscovery.class).getExtension(ServiceDiscoveryEnum.ZK.getName());
        this.unprocessedRequests = SingletonFactory.getInstance(UnprocessedRequests.class);
        this.channelProvider = SingletonFactory.getInstance(ChannelProvider.class);
    }

当数据发出时，会被经过的一个个ChannelOutboundHandler（简称out）依次处理，而不会被ChannelInboundHandler（简称in）处理， RpcMessageEncoder对应的是out，因为写出时才会需要编码，RpcMessageDecoder和NettyRpcClientHandler对应的是in，因为读取服务端数据时才需要解码以及处理响应数据。

服务端也是类似：

   try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    // TCP默认开启了 Nagle 算法，该算法的作用是尽可能的发送大数据快，减少网络传输。TCP_NODELAY 参数的作用就是控制是否启用 Nagle 算法。
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    // 是否开启 TCP 底层心跳机制
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    //表示系统用于临时存放已完成三次握手的请求的队列的最大长度,如果连接建立频繁，服务器处理创建新连接较慢，可以适当调大这个参数
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    // 当客户端第一次进行请求的时候才会进行初始化
                    .childHandler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            // 30 秒之内没有收到客户端请求的话就关闭连接
                            ChannelPipeline p = ch.pipeline();
                            p.addLast(new IdleStateHandler(30, 0, 0, TimeUnit.SECONDS));
                            p.addLast(new RpcMessageEncoder());
                            p.addLast(new RpcMessageDecoder());
                            p.addLast(serviceHandlerGroup, new NettyRpcServerHandler());
                        }
                    });
            // 绑定端口，同步等待绑定成功
            ChannelFuture f = b.bind(host, PORT).sync();
            // 等待服务端监听端口关闭
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("occur exception when start server:", e);
        } finally {
            log.error("shutdown bossGroup and workerGroup");
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            serviceHandlerGroup.shutdownGracefully();
        }

服务端也使用RpcMessageEncoder和RpcMessageDecoder来进行编解码，因为整个框架的编解码需要使用同一套规则。NettyRpcServerHandler自然对应的是in，因为需要在接收读取数据时进行处理，服务端具体通过反射调用执行方法获取执行结果的过程就发生在这里。

除此之外，网络通信还需要解决粘包半包的问题，也需要指定一套通信时的规则并可以被双方感知。因此需要自定义rpc协议，这也是rpc与http的区别之一。这里给出本RPC框架自定义的数据格式：