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 类结构图
其中，AbstractConfigurableInstance 是可配置实例抽象类，包含配置容器和全局开关（在《Config 配置设计》中进行分析）；
AbstractRemotingServer 使用
模板模式实现了 父类 RemotingServer 的方法，并提供了三个 doXxx() 方法由子类来覆盖；
RpcServer 中前6个属性除了 userProcessors 都是 netty 相关；后三个与 Connection 相关；RpcRemoting 是调用客户端的工具类。
注意：在 RemotingServer 中有void registerDefaultExecutor(byte protocolCode, ExecutorService executor)，在《线程池设计》部分进行分析。
