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

注意：invokeWithFuture 第三个参数 int timeoutMillis 指的是 io.netty.util.TimerTask 延迟启动时间；而 future 阻塞等待的超时时间 是在future.get( int timeoutMillis) 中进行设置的。

一、线程模型图

image.png

总结：

future 与 sync 只在“发出请求后是否阻塞等待”处不同，其他相同（包括阻塞点都是 InvokeFuture.waitResponse() 阻塞在一个 CountDownLatch 上）

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

二、代码执行流程梯形图

2.1 客户端发出请求

-->RpcClient.invokeWithFuture(String addr, Object request, int timeoutMillis)
  -->RpcClientRemoting.invokeWithFuture(String addr, Object request, InvokeContext invokeContext, int timeoutMillis) // invokeContext = null
    -->Url url = this.addressParser.parse(addr) // 将 addr 转化为 Url
    -->RpcClientRemoting.invokeWithFuture(Url url, Object request, InvokeContext invokeContext, int timeoutMillis)
      <!-- 一、获取连接 -->
      -->Connection conn = getConnectionAndInitInvokeContext(url, invokeContext)
      <!-- 二、检查连接 -->
      -->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.invokeWithFuture(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 创建Timeout实例并设置到future实例中 -->
        -->TimerHolder.getTimer().newTimeout
        -->InvokeFuture.addTimeout(Timeout timeout)
        <!-- 4.3 使用 netty 发送消息 -->
        -->conn.getChannel().writeAndFlush(request) // netty发送消息 然后返回刚刚创建的InvokeFuture实例
        <!-- 4.4 创建RpcResponseFuture，包装InvokeFuture实例 -->
        -->new RpcResponseFuture(String addr, InvokeFuture future)

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

总结：

1. 获取连接

2. 检查连接

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

4. 发起请求

• 创建 InvokeFuture，并将 {invokeId : InvokeFuture实例} 存储到 Connection 的Map<Integer, InvokeFuture> invokeFutureMap
• 创建 io.netty.util.Timeout 实例并设置到future实例中
• 使用 netty 发送消息
• 创建RpcResponseFuture，包装InvokeFuture实例

2.2 客户端阻塞等待响应

RpcResponseFuture.get()
<!-- 一、当前线程阻塞在这里，等待响应返回并填充到future后，再进行唤醒 -->
-->RemotingCommand response = InvokeFuture.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() // 抽取响应结果

总结：

1. 如果服务端已经填充好了相应数据，则直接进行第2步对响应消息进行处理；否则，当前线程阻塞（countDownLatch.await(timeoutMillis)），等待响应返回并填充到 InvokeFuture 后，再进行唤醒（countDownLatch.countDown()）

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

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

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

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

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发送响应

总结：

与 Sync 完全相同

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.4 客户端接收响应

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 填充响应 + 唤醒主线程 + 取消 io.netty.util.TimerTask 超时任务 + 如果有回调，调用回调 -->
      -->InvokeFuture.putResponse(RemotingCommand response)
        -->this.countDownLatch.countDown() // 解锁等待，唤醒主线程
      -->future.cancelTimeout();
      -->DefaultInvokeFuture.executeInvokeCallback() // 执行相应的 callbackListener

总结：

唯一与 sync 不同的是对于 Future 和 Callback 模式，启动了 netty 的时间轮超时任务，此处会取消该 TimerTask 任务

1. 创建 InvokeContext 和 RemotingContext

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

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

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

• 从连接中根据响应 id 获取请求的 InvokeFuture
• 填充响应 + 唤醒阻塞线程 + 取消 io.netty.util.TimerTask 超时任务 + 如果有回调，调用回调