MyRequest request = new MyRequest();
 request.setReq("hello, bolt-server");
 MyResponse response = (MyResponse) client.invokeSync("127.0.0.1:8888", request, 30 * 1000);

注意：invokeSync 第三个参数 int timeoutMillis 指的是 future 阻塞等待的超时时间；连接超时时间由 RpcConfigs.CONNECT_TIMEOUT_KEY 来指定。

一、线程模型图

image.png

总结：

1. 客户端用户线程 user-thread 发出请求（实际上是将 netty task 压到 netty 处理队列中，netty 客户端 worker 线程进行真正的请求发出），然后阻塞等待响应
2. 服务端 worker 线程接收请求，根据是否在 IO 线程执行所有操作来决定是否使用一个 Bolt 线程池（或者自定义的线程池）来处理业务
3. 服务端返回响应后，客户端 worker 线程接收到响应，将响应转发给 Bolt 线程池（或者自定义的线程池）
4. Bolt 线程池（或者自定义的线程池）中的线程将响应设置到相应的 InvokeFuture 中，之后唤醒阻塞的 user-thread
5. user-thread 进行反序列化和响应的抽取，最后返回给调用处

二、代码执行流程梯形图

2.1 客户端发出请求

-->RpcClient.invokeSync(String addr, Object request, int timeoutMillis)
  -->RpcClientRemoting.invokeSync(String addr, Object request, InvokeContext invokeContext, int timeoutMillis) // invokeContext = null
    -->Url url = this.addressParser.parse(addr) // 将 addr 转化为 Url
    -->RpcClientRemoting.invokeSync(Url url, Object request, InvokeContext invokeContext, int timeoutMillis)
      <!-- 一、获取连接 -->
      -->Connection conn = getConnectionAndInitInvokeContext(url, invokeContext)
        -->DefaultConnectionManager.getAndCreateIfAbsent(Url url)
          -->new ConnectionPoolCall(Url url) // 创建一个 Callable<ConnectionPool> task
          <!-- 1.1 获取或者创建 poolKey 的 ConnectionPool -->
          -->pool = getConnectionPoolAndCreateIfAbsent(String poolKey, Callable<ConnectionPool> callable) // poolkey: 127.0.0.1:8999 callable: 上述的 ConnectionPoolCall 实例
            -->initialTask = new RunStateRecordedFutureTask<ConnectionPool>(callable) // 包装 callable
            -->Map<String, RunStateRecordedFutureTask<ConnectionPool>> connTasks.putIfAbsent(poolKey, initialTask) //下一次就直接从该缓存取出任务 initialTask（之后直接从 task 中取出 ConnectionPool），不再 new
            -->initialTask.run()
              -->ConnectionPoolCall.call()
                <!-- 1.1.1 获取或者创建 poolKey 的 ConnectionPool -->
                -->pool = new ConnectionPool(connectionSelectStrategy)
                <!-- 1.1.2 创建 poolKey 的 Connection 并添加到 ConnectionPool 中 -->
                -->DefaultConnectionManager.doCreate(Url url, ConnectionPool pool, String taskName, int syncCreateNumWhenNotWarmup)
                  -->Connection connection = create(url)
                    -->RpcConnectionFactory.createConnection(Url url)
                      -->doCreateConnection(String targetIP, int targetPort, int connectTimeout)
                      -->bootstrap.connect(new InetSocketAddress(targetIP, targetPort)) // netty 客户端连接创建服务端
                    -->new Connection(Channel channel, ProtocolCode protocolCode, byte version, Url url) // 包装 Channel，并且初始化了一堆 attr 附加属性
                    -->channel.pipeline().fireUserEventTriggered(ConnectionEventType.CONNECT) // 触发连接事件（此时 ConnectionEventHandler 就会调用 ConnectionEventListener 执行其内的 List<ConnectionEventProcessor> 的 onEvent 方法）
                  -->pool.add(connection)
          <!-- 1.2 从 ConnectionPool 中获取 connection 连接 -->
          -->pool.get() // 从 ConnectionPool 中使用 ConnectionSelectStrategy 获取一个 Connection
      
      <!-- 二、检查连接 -->
      -->this.connectionManager.check(conn) // 校验 connection 不为 null && channel 不为 null && channel 是 active 状态 && channel 可写
      <!-- 三、创建请求对象 -->
      -->RpcCommandFactory.createRequestCommand(Object requestObject)
        -->new RpcRequestCommand(Object request) // 设置唯一id + 消息类型为 Request（还有 Response 和 heartbeat）+ MyRequest request
        -->command.serialize() // 序列化
      <!-- 四、发起请求 -->
      -->BaseRemoting.invokeSync(Connection conn, RemotingCommand request, int timeoutMillis)
        <!-- 4.1 创建InvokeFuture，并将 { invokeId : InvokeFuture实例} 存储到 Connection 的 Map<Integer, InvokeFuture> invokeFutureMap -->
        -->InvokeFuture future = new DefaultInvokeFuture(int invokeId, InvokeCallbackListener callbackListener, InvokeCallback callback, byte protocol, CommandFactory commandFactory, InvokeContext invokeContext);
        -->Connection.addInvokeFuture(InvokeFuture future)
          -->Connection.(Map<Integer, InvokeFuture> invokeFutureMap).putIfAbsent(future.invokeId(), future) // future.invokeId()就是消息的唯一id，后续的响应也会塞入这个id，最后根据响应中的该id来获取对应的InvokeFuture，做相应的操作
        <!-- 4.2 使用 netty 发送消息 -->
        -->conn.getChannel().writeAndFlush(request) // netty发送消息
        <!-- 4.3 当前线程阻塞在这里，等待响应返回并填充到future后，再进行唤醒 -->
        -->RemotingCommand response = future.waitResponse(timeoutMillis) //阻塞等待消息
          -->this.countDownLatch.await(timeoutMillis, TimeUnit.MILLISECONDS)

      ... 服务端处理并返回响应 ...
      ... 客户端 netty worker 线程接收响应并填充到指定 invokeId 的 InvokeFuture 中，唤醒如下流程 ...

      RpcResponseResolver.resolveResponseObject(ResponseCommand responseCommand, String addr)
      -->preProcess(ResponseCommand responseCommand, String addr) // 处理错误的响应，如果有，直接封装为响应的 Exception，然后 throw
      -->toResponseObject(ResponseCommand responseCommand) // 如果是成功状态
      -->response.deserialize() // 反序列化
      -->response.getResponseObject() // 抽取响应结果

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

总结：

1. 获取连接

2. 检查连接

3. 使用 RpcCommandFactory 创建请求对象 + 序列化

4. 发起请求

• 创建 InvokeFuture，并将 {invokeId : InvokeFuture实例} 存储到 Connection 的Map<Integer, InvokeFuture> invokeFutureMap
• 使用 netty 发送消息
• 当前线程阻塞（countDownLatch.await(timeoutMillis)），等待响应返回并填充到future后，再进行唤醒（countDownLatch.countDown()）

服务端处理请求并返回响应

客户端 netty worker 线程接收响应并填充到指定 invokeId 的 InvokeFuture 中，唤醒当前线程

5. 首先处理错误响应的情况，如果响应是错误状态，直接封装为响应的 Exception，然后 throw；如果响应是成功状态，进行 RpcResponseCommand 的反序列化，之后从 RpcResponseCommand 抽取真正的相应数据

2.2 服务端处理请求并返回响应

RpcHandler.channelRead(ChannelHandlerContext ctx, Object msg)
<!-- 一、创建上下文 -->
-->new InvokeContext()
-->new RemotingContext(ChannelHandlerContext ctx, InvokeContext invokeContext, boolean serverSide, ConcurrentHashMap<String, UserProcessor<?>> userProcessors)
<!-- 二、根据 channel 中的附加属性获取相应的 Protocol，之后使用该 Protocol 实例的 CommandHandler 处理消息 -->
-->RpcCommandHandler.handle(RemotingContext ctx, Object msg)
  <!-- 2.1 从 CommandHandler 中获取 CommandCode 为 REQUEST 的 RemotingProcessor 实例 RpcRequestProcessor，之后使用 RpcRequestProcessor 进行请求处理-->
  -->RpcRequestProcessor.process(RemotingContext ctx, RpcRequestCommand cmd, ExecutorService defaultExecutor)
    <!-- 2.1.1 反序列化clazz（感兴趣key），用于获取相应的UserProcessor；如果相应的UserProcessor==null，创建异常响应，发送给调用端，否则，继续执行 -->
    -->反序列化 clazz
    <!-- 
        2.1.2 如果 userProcessor.processInIOThread()==true，直接对请求进行反序列化，然后创建ProcessTask任务，最后直接在当前的netty worker线程中执行ProcessTask.run()；
              否则，如果用户自定义了ExecutorSelector，则从众多的自定义线程池选择一个线程池，如果没定义，则使用自定义的线程池userProcessor.getExecutor()，如果最后没有自定义的线程池，则使用ProcessorManager的defaultExecutor，
              来执行ProcessTask.run()
    -->
      <!-- ProcessTask.run() -->
      -->RpcRequestProcessor.doProcess(RemotingContext ctx, RpcRequestCommand cmd)
        -->反序列化header、content
        -->dispatchToUserProcessor(RemotingContext ctx, RpcRequestCommand cmd)
          <!-- 构造用户业务上下文 -->
          -->new DefaultBizContext(remotingCtx)
          <!-- 使用用户自定义处理器处理请求 -->
          -->MyServerUserProcessor.handleRequest(BizContext bizCtx, MyRequest request)
          <!-- 创建响应 -->
          -->RemotingCommand response = RpcCommandFactory.createResponse(Object responseObject, RemotingCommand requestCmd) // 这里将response.id = requestCmd.id
          <!-- 序列化响应并发送 -->
          -->RpcRequestProcessor.sendResponseIfNecessary(RemotingContext ctx, byte type, RemotingCommand response)
            -->response.serialize() // 序列化
            -->ctx.writeAndFlush(serializedResponse) // netty发送响应

总结：

1. 创建 InvokeContext 和 RemotingContext

2. 根据 channel 中的附加属性获取相应的 Protocol，之后使用该 Protocol 实例的 CommandHandler 处理消息

• 从 CommandHandler 中获取 CommandCode 为 REQUEST 的 RemotingProcessor 实例 RpcRequestProcessor，之后使用 RpcRequestProcessor 进行请求处理

• 反序列化clazz（感兴趣key），用于获取相应的UserProcessor；如果相应的 UserProcessor==null，创建异常响应，发送给调用端，否则，继续执行
• 如果 userProcessor.processInIOThread()==true，直接对请求进行反序列化，然后创建 ProcessTask 任务，最后直接在当前的 netty worker 线程中执行 ProcessTask.run()；
  否则，如果用户 UserProcessor 自定义了 ExecutorSelector，则从众多的自定义线程池选择一个线程池，如果没定义，则使用 UserProcessor 自定义的线程池 userProcessor.getExecutor()，如果还没有，则使用 RemotingProcessor 自定义的线程池 executor，如果最后没有自定义的线程池，则使用 ProcessorManager 的defaultExecutor，来执行ProcessTask.run()

• 反序列化 header、content（如果用户自定义了 ExecutorSelector，则header的反序列化需要提前，header 会作为众多自定义线程池的选择参数）
• 构造用户业务上下文 DefaultBizContext
• 使用用户自定义处理器处理请求
• 创建响应，序列化响应并发送

2.3 客户端接收响应

RpcHandler.channelRead(ChannelHandlerContext ctx, Object msg)
<!-- 一、创建上下文 -->
-->new RemotingContext(ctx, new InvokeContext(), serverSide, userProcessors)
<!-- 二、根据 channel 中的附加属性获取相应的 Protocol，之后使用该 Protocol 实例的 CommandHandler 处理消息 -->
-->RpcCommandHandler.handle(RemotingContext ctx, Object msg)
  <!-- 三、获取线程池（如果 RemotingProcessor 自定义了线程池 executor 执行 ProcessTask.run()，否则使用 ProcessorManager 的 defaultExecutor）-->
  -->AbstractRemotingProcessor.process(RemotingContext ctx, T msg, ExecutorService defaultExecutor)
    -->new ProcessTask(RemotingContext ctx, T msg)
    <!-- ProcessTask.run() -->
    -->RpcResponseProcessor.doProcess(RemotingContext ctx, RemotingCommand cmd) // cmd是RpcResponseCommand
      <!-- 3.1 从连接中根据响应id获取请求的 InvokeFuture -->
      -->InvokeFuture future = conn.removeInvokeFuture(cmd.getId()) // 根据响应id获取请求的 InvokeFuture
      <!-- 3.2 填充响应 + 唤醒主线程 + 如果有回调，调用回调 -->
      -->InvokeFuture.putResponse(RemotingCommand response)
        -->this.countDownLatch.countDown() // 解锁等待，唤醒主线程
      -->DefaultInvokeFuture.executeInvokeCallback() // 执行相应的 callbackListener

总结：

1. 创建 InvokeContext 和 RemotingContext

2. 根据 channel 中的附加属性获取相应的 Protocol，之后使用该 Protocol 实例的 CommandHandler 处理消息

• 从 CommandHandler 中获取 CommandCode 为 RESPONSE 的 RemotingProcessor 实例 RpcResponseProcessor，之后使用 RpcResponseProcessor 进行响应处理

• 如果RemotingProcessor自定义了线程池executor执行ProcessTask.run()，否则使用ProcessorManager的defaultExecutor
  ProcessTask.run()：

• 从连接中根据响应 id 获取请求的 InvokeFuture
• 填充响应 + 唤醒阻塞线程 + 如果有回调，调用回调

关于自定义线程池与默认线程池的设计和使用，在《线程池设计》中分析。
关于三种上下文的设计与使用，在《上下文设计》中分析。