fescar源码分析-AbstractRpcRemotingSe

在fescar源码分析-AbstractRpcRemoting介绍了fescar对响应请求及向送请求的封装。下面分析AbstractRpcRemoting的子类AbstractRpcRemotingServer。

image.png

AbstractRpcRemotingServer在父类基础针对RPC调用的服务端做进一步的封装，实现RPC调用服务端的实例创建。

1.通过构造函数初始化RPC服务端的相关配置。
2.start配置RPC服务端并启动服务。
3.shutdown关闭RPC服务端服务。

AbstractRpcRemotingServer是使用Netty的ServerBootstrap 类负责引导创建一个RPC服务端。下图展示了如何工作：

Figure%209

• 1.当调用 bind() 后 ServerBootstrap 将创建一个新的管道，这个管道将会在绑定成功后接收子管道
• 2.接收新连接给每个子管道
• 3.接收连接的 Channel

要使用Netty创建PRC的客户端，需要实现以下逻辑:

ServerBootstrap bootstrap = new ServerBootstrap(); //1
bootstrap.group(new NioEventLoopGroup(),new NioEventLoopGroup()) //2
    .channel(NioServerSocketChannel.class) //3
    .childHandler(new SimpleChannelInboundHandler<ByteBuf>() { //4
        @Override
        protected void channelRead0(ChannelHandlerContext ctx,
            ByteBuf byteBuf) throws Exception {
                System.out.println("Reveived data");
                byteBuf.clear();
            }
        }
    );
ChannelFuture future = bootstrap.bind(new InetSocketAddress(8080)); //5

• 1.创建要给新的 ServerBootstrap 来创建新的 SocketChannel 管道并绑定他们。
• 2.指定 BossEventLoopGroup 和WorkerEventLoopGroup 用于从注册的 ServerChannel 中获取EventLoop 和接收到的管道。
• 3.指定要使用的管道类
• 4.设置子处理器用于处理接收的管道的 I/O 和数据
• 5.通过配置引导来绑定端口

如下图所示：

image

下面看一下AbstractRpcRemotingServer的具体实现:

创建ServerBootstrap

在AbstractRpcRemotingServer类中定义了一个ServerBootstrap成员属性，并在构造函数中初始化。

private final ServerBootstrap serverBootstrap;
......
public AbstractRpcRemotingServer(final NettyServerConfig nettyServerConfig,
                                     final ThreadPoolExecutor messageExecutor, final ChannelHandler... handlers) {
        super(messageExecutor);
        this.serverBootstrap = new ServerBootstrap();

指定 EventLoopGroup

在创建函数中初始化eventLoopGroupBoss和eventLoopGroupWorker，如果本地环境支持Epoll,则使用EpollEventLoopGroup创建实例。

        if (NettyServerConfig.enableEpoll()) {
            this.eventLoopGroupBoss = new EpollEventLoopGroup(nettyServerConfig.getBossThreadSize(),
                new NamedThreadFactory(nettyServerConfig.getBossThreadPrefix(), nettyServerConfig.getBossThreadSize()));
            this.eventLoopGroupWorker = new EpollEventLoopGroup(nettyServerConfig.getServerWorkerThreads(),
                new NamedThreadFactory(nettyServerConfig.getWorkerThreadPrefix(),
                    nettyServerConfig.getServerWorkerThreads()));
        }

否则使用NioEventLoopGroup创建实例。

            this.eventLoopGroupBoss = new NioEventLoopGroup(nettyServerConfig.getBossThreadSize(),
                new NamedThreadFactory(nettyServerConfig.getBossThreadPrefix(), nettyServerConfig.getBossThreadSize()));
            this.eventLoopGroupWorker = new NioEventLoopGroup(nettyServerConfig.getServerWorkerThreads(),
                new NamedThreadFactory(nettyServerConfig.getWorkerThreadPrefix(),
                    nettyServerConfig.getServerWorkerThreads()));

不管是NioEventLoopGroup还是EpollEventLoopGroup，都使用配置的参数初始化EventLoopGroup的线程数及线程名称等。然后在启动RPC服务端时，将EventLoopGroup实例设置在EventLoopGroup中。

this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupWorker)

指定Channel

在start()方法中将根据配置文件及默认参数设置服务端的通道类。

.channel(nettyServerConfig.SERVER_CHANNEL_CLAZZ)

设置服务端属性

            .option(ChannelOption.SO_BACKLOG, nettyServerConfig.getSoBackLogSize())
            .option(ChannelOption.SO_REUSEADDR, true)
            .childOption(ChannelOption.SO_KEEPALIVE, true)
            .childOption(ChannelOption.TCP_NODELAY, true)
            .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSendBufSize())
            .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketResvBufSize())
            .childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK,
                nettyServerConfig.getWriteBufferHighWaterMark())
            .childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, nettyServerConfig.getWriteBufferLowWaterMark())
......
        if (nettyServerConfig.isEnableServerPooledByteBufAllocator()) {
            this.serverBootstrap.childOption(ChannelOption.ALLOCATOR, NettyServerConfig.DIRECT_BYTE_BUF_ALLOCATOR);
        }

设置Handler

初始化eventLoopGroupWorker的Handler。

            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) {
                    ch.pipeline().addLast(new IdleStateHandler(nettyServerConfig.getChannelMaxReadIdleSeconds(), 0, 0))//1
                        .addLast(new MessageCodecHandler());//2
                    if (null != channelHandlers) {
                        addChannelPipelineLast(ch, channelHandlers);//3
                    }
                }
            });

• 1.在通道中添加IdleStateHandler。
• 2.在通道中添加MessageCodecHandler，用于数据的编码及解码。
• 3.在通道中添加子类定义的Handler。

绑定端口

服务引导类绑定指定的端口，并启动RPC服务端。

        try {
            ChannelFuture future = this.serverBootstrap.bind(listenPort).sync();
            LOGGER.info("Server started ... ");
            future.channel().closeFuture().sync();
        } catch (InterruptedException exx) {
            throw new RuntimeException(exx);
        }