SOFABolt 源码分析2 - RpcServer 服务端启动

作者: 原水寒 | 来源:发表于2018-10-04 14:40 被阅读301次

 RpcServer server = new RpcServer(8888);
 server.registerUserProcessor(new MyServerUserProcessor());
 server.start();

一、代码执行流程梯形图

<!-- 一、创建 RpcServer -->
new RpcServer(port)
-->EventLoopGroup workerGroup
-->EventLoopGroup bossGroup
-->new RpcCodec
-->new ConnectionEventListener()
-->new ConcurrentHashMap<String, UserProcessor<?>> userProcessors

<!-- 二、添加用户自定义业务逻辑处理器 -->
RpcServer.registerUserProcessor(UserProcessor<?> processor)

<!-- 三、初始化并启动 RpcServer 实例 -->
AbstractRemotingServer.start()
-->RpcServer.doInit()
  <!-- 3.1 关联连接事件处理器与连接事件监听器，并由连接事件处理器中的连接执行器来执行连接监听器中的 processor -->
  -->new ConnectionEventHandler(GlobalSwitch globalSwitch) // 连接事件处理器
  -->ConnectionEventHandler.setConnectionEventListener(ConnectionEventListener listener) // 关联连接事件处理器与连接事件监听器
  -->new ConnectionEventExecutor() // 创建连接事件执行器（后续 ConnectionEventListener 中的 ConnectionEventProcessor 的执行都由该线程池来完成）
  <!-- 3.2 创建真正的请求执行客户端（发起调用类） -->
  -->initRpcRemoting()
    <!-- 3.2.1 创建两种协议实例 + 注册到 RpcProtocolManager -->
    -->RpcProtocolManager.initProtocols()
      -->new RpcProtocol()
        -->this.encoder = new RpcCommandEncoder(); // 真正的最底层的编码器
        -->this.decoder = new RpcCommandDecoder(); // 真正的最底层的解码器
        -->this.commandFactory = new RpcCommandFactory(); // 创建请求信息和返回信息包装体的工厂
        -->this.heartbeatTrigger = new RpcHeartbeatTrigger(this.commandFactory); // 最底层的连接心跳处理器
        -->this.commandHandler = new RpcCommandHandler(this.commandFactory); 
          -->this.processorManager = new ProcessorManager() // 创建 processor 管理器
            -->Map<CommandCode, RemotingProcessor<?>> cmd2processors
            -->ExecutorService defaultExecutor
          -->this.processorManager.registerProcessor(RpcCommandCode.RPC_REQUEST, new RpcRequestProcessor(this.commandFactory))
          -->this.processorManager.registerProcessor(RpcCommandCode.RPC_RESPONSE, new RpcResponseProcessor())
          -->this.processorManager.registerProcessor(CommonCommandCode.HEARTBEAT, new RpcHeartBeatProcessor());
          -->this.processorManager.registerDefaultProcessor(直接 logger.error)
      -->RpcProtocolManager.registerProtocol(Protocol protocol, byte... protocolCodeBytes) // 将 RpcProtocol 实例添加到 RpcProtocolManager 的 Map<ProtocolCode, Protocol> protocols 中
      -->new RpcProtocol2()
        -->this.encoder = new RpcCommandEncoderV2();
        -->this.decoder = new RpcCommandDecoderV2();
        -->this.commandFactory = new RpcCommandFactory();
        -->this.heartbeatTrigger = new RpcHeartbeatTrigger(this.commandFactory);
        -->this.commandHandler = new RpcCommandHandler(this.commandFactory);
      -->RpcProtocolManager.registerProtocol(Protocol protocol, byte... protocolCodeBytes) // 将 RpcProtocolV2 实例添加到 RpcProtocolManager 的 Map<ProtocolCode, Protocol> protocols 中
    <!-- 3.2.2 创建请求信息和返回信息包装体的工厂 -->
    -->new RpcCommandFactory()
    <!-- 3.2.3 创建 RpcServerRemoting (发起底层调用实现类) -->
    -->new RpcServerRemoting(CommandFactory commandFactory, RemotingAddressParser addressParser, DefaultConnectionManager connectionManager)
  <!-- 3.3 配置 netty 服务端 -->
  -->new ServerBootstrap() 并做一系列配置
  // netty 业务逻辑处理器
  -->new RpcHandler(boolean serverSide, ConcurrentHashMap<String, UserProcessor<?>> userProcessors) // {"com.alipay.remoting.mydemo.MyRequest" : MyServerUserProcessor}
  -->netty 连接 channel 创建成功后，将 channel 包装为 Connection 对象
-->RpcServer.doStart()
  -->this.bootstrap.bind(new InetSocketAddress(ip(), port())).sync()

总结：

关于连接 Connection 相关的，放在《Connection 连接设计》章节分析，此处跳过；
关于心跳 HeartBeat 相关的，放在《HeartBeat 心跳设计》章节分析，此处跳过。

创建 RpcServer 实例

创建 workerGroup（static类变量，实现多个 RpcServer 实例共享 workerGroup）与 bossGroup

创建 Codec 的实现类 RpcCodec 实例，用于创建 netty 的编解码器，实质上是一个工厂类

创建用户处理器 UserProcessor 实现类容器 Map<String, UserProcessor<?>> userProcessors

添加 UserProcessor 实现类到 userProcessors

{ "感兴趣的请求数据类型" ：UserProcessor实现类 }
eg. key = "com.alipay.remoting.mydemo.MyRequest"，value = MyServerUserProcessor 实例

初始化并启动 RpcServer 实例

初始化 RpcServer

创建两种协议 RpcProtocol 和 RpcProtocolV2 实例，添加到 RpcProtocolManager 的 Map<ProtocolCode, Protocol> protocols 协议容器中；每一种协议都有以下5个属性：

CommandEncoder：编码器实例
CommandDecoder：解码器实例
HeartbeatTrigger：心跳触发器实例
CommandFactory：创建 Remote 层请求和响应封装实体的创建工厂实例
CommandHandler：Remote 层的消息处理器，包含一个 ProcessorManager 实例，其包含ConcurrentHashMap<CommandCode, RemotingProcessor<?>> cmd2processors容器，存储了三个 Processor 处理器

请求消息处理器：{ RpcCommandCode.RPC_REQUEST : RpcRequestProcessor 实例 }
响应消息处理器：{ RpcCommandCode.RPC_RESPONSE : RpcResponseProcessor 实例 }
心跳消息处理器（心跳发送与心跳响应消息）：{ CommonCommandCode.HEARTBEAT : RpcHeartBeatProcessor 实例 }
默认处理器：AbstractRemotingProcessor 匿名内部类，直接 logger.error。该处理器用于娄底，当要处理的消息不是上述三种的任何一个，则使用该处理器

创建 Remote 层请求和响应封装实体的创建工厂 RpcCommandFactory 实例

创建 RpcServerRemoting (发起底层调用实现类) 实例

SOFABolt 可以进行双向调用，server 端也可以调用 client 端，所以此处构建了 RpcServerRemoting 实例

创建 ServerBootstrap 实例并设置一系列 netty 服务端配置

创建 RpcHandler 实例作为 netty 的业务逻辑处理器，之后会有一个 Processor 处理链进行处理

请求链 RpcHandler -> RpcCommandHandler -> RpcRequestProcessor -> UserProcessor
响应链 RpcHandler -> RpcCommandHandler -> RpcResponseProcessor
心跳链 RpcHandler -> RpcCommandHandler -> RpcHeartBeatProcessor

启动 RpcServer

this.bootstrap.bind

二、从 netty 的角度看执行链

        this.bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel channel) {
                ChannelPipeline pipeline = channel.pipeline();
                pipeline.addLast("decoder", codec.newDecoder());
                pipeline.addLast("encoder", codec.newEncoder());
                if (idleSwitch) {
                    pipeline.addLast("idleStateHandler", new IdleStateHandler(0, 0, idleTime,
                        TimeUnit.MILLISECONDS));
                    pipeline.addLast("serverIdleHandler", serverIdleHandler);
                }
                pipeline.addLast("connectionEventHandler", connectionEventHandler);
                pipeline.addLast("handler", rpcHandler);
                createConnection(channel);
            }
        }

跳过心跳 HeartBeat 逻辑、跳过连接 Connection 逻辑，我们只看编解码器和业务处理器逻辑。
这里大概给出编解码链，真正的编解码过程在《Codec 编解码设计》的时候再详细分析。
这里大概给出调用链，真正的调用处理过程在分析四种调用模式设计的时候再详细分析。

2.1 编解码器 RpcCodec

编码链 ProtocolCodeBasedEncoder -> CommandEncoder
解码链 RpcProtocolDecoder -> CommandDecoder

-------------------------------- netty 编解码器 构造工厂 ------------------------------------------
public class RpcCodec implements Codec {
    @Override
    public ChannelHandler newEncoder() {
        return new ProtocolCodeBasedEncoder(ProtocolCode.fromBytes(RpcProtocolV2.PROTOCOL_CODE));
    }
    @Override
    public ChannelHandler newDecoder() {
        return new RpcProtocolDecoder(RpcProtocolManager.DEFAULT_PROTOCOL_CODE_LENGTH);
    }
}

--------------------------- netty 编码器：CommandEncoder的包装类 -----------------------------------
public class ProtocolCodeBasedEncoder extends MessageToByteEncoder<Serializable> {
    ...

    @Override
    protected void encode(ChannelHandlerContext ctx, Serializable msg, ByteBuf out)
                                                                                   throws Exception {
        Attribute<ProtocolCode> att = ctx.channel().attr(Connection.PROTOCOL);
        ...
        Protocol protocol = ProtocolManager.getProtocol(protocolCode);
        // 使用protocol的编码器进行编码
        protocol.getEncoder().encode(ctx, msg, out);
    }
}

--------------------------- netty 解码器：CommandDecoder的包装类 -----------------------------------
public class RpcProtocolDecoder extends ProtocolCodeBasedDecoder {
    ...

    @Override
    protected byte decodeProtocolVersion(ByteBuf in) {
       ...
    }
}

public class ProtocolCodeBasedDecoder extends AbstractBatchDecoder {
    ...

   protected byte decodeProtocolVersion(ByteBuf in) {
        ...
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        in.markReaderIndex();
        ProtocolCode protocolCode = decodeProtocolCode(in);
        if (null != protocolCode) {
            byte protocolVersion = decodeProtocolVersion(in);
            ...
            Protocol protocol = ProtocolManager.getProtocol(protocolCode);
            if (null != protocol) {
                in.resetReaderIndex();
                // 使用protocol的解码器进行解码
                protocol.getDecoder().decode(ctx, in, out);
            } else {
                throw new CodecException("Unknown protocol code: [" + protocolCode
                                         + "] while decode in ProtocolDecoder.");
            }
        }
    }
}

2.2 业务处理器 RpcHandler

请求链 RpcHandler -> RpcCommandHandler -> RpcRequestProcessor -> UserProcessor
响应链 RpcHandler -> RpcCommandHandler -> RpcResponseProcessor
心跳链 RpcHandler -> RpcCommandHandler -> RpcHeartBeatProcessor

----------------- RpcHandler.channelRead(ChannelHandlerContext ctx, Object msg) -------------
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 每一个请求都会在连接上添加 ProtocolCode 属性
        ProtocolCode protocolCode = ctx.channel().attr(Connection.PROTOCOL).get();
        // 使用 ProtocolCode 获取 Protocol
        Protocol protocol = ProtocolManager.getProtocol(protocolCode);
        // 使用 Protocol 获取其 RpcCommandHandler 实例，使用 RpcCommandHandler 实例进行消息处理
        protocol.getCommandHandler().handleCommand(
            new RemotingContext(ctx, new InvokeContext(), serverSide, userProcessors), msg);
    }

三、RpcServer 类结构图

image.png

其中，AbstractConfigurableInstance 是可配置实例抽象类，包含配置容器和全局开关（在《Config 配置设计》中进行分析）；

AbstractRemotingServer 使用 模板模式 实现了父类 RemotingServer 的方法，并提供了三个 doXxx() 方法由子类来覆盖；
RpcServer 中前6个属性除了 userProcessors 都是 netty 相关；后三个与 Connection 相关；RpcRemoting 是调用客户端的工具类。
注意：在 RemotingServer 中有 void registerDefaultExecutor(byte protocolCode, ExecutorService executor)，在《线程池设计》部分进行分析。

SOFABolt 源码分析2 - RpcServer 服务端启动

一、代码执行流程梯形图

总结：

二、从 netty 的角度看执行链

2.1 编解码器 RpcCodec

2.2 业务处理器 RpcHandler

三、RpcServer 类结构图

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

sofabolt