SOFABolt 源码分析7 - Callback 异步通信方式的设计

 MyRequest request = new MyRequest();
 request.setReq("hello, bolt-server");
 client.invokeWithCallback("127.0.0.1:8888", request, new InvokeCallback() {
       @Override
        public void onResponse(Object result) {
            System.out.println("回调成功：" + (MyResponse) result);
        }
       @Override
        public void onException(Throwable e) {
            System.out.println("回调异常：" + e.getMessage());
        }
        @Override
        public Executor getExecutor() {
            return null;
        }
    }, 30 * 1000);

注意：invokeWithCallback 第四个参数 int timeoutMillis 指的是 io.netty.util.TimerTask 延迟启动时间；
invokeWithCallback 第三个参数 就是回调函数 InvokeCallback 实现类。

一、线程模型图

image.png

总结：

  1. 客户端用户线程 user-thread 发出请求（实际上是将 netty task 压到 netty 处理队列中，netty 客户端 worker 线程进行真正的请求发出），之后 user-thread 就可以去做其他事了
  2. 服务端 worker 线程接收请求，根据是否在 IO 线程执行所有操作来决定是否使用一个 Bolt 线程池（或者自定义的线程池）来处理业务
  3. 服务端返回响应后，客户端 worker 线程接收到响应，将响应转发给 Bolt 线程池（或者自定义的线程池）
  4. Bolt 线程池（或者自定义的线程池）中的线程将响应设置到相应的 InvokeFuture 中，取消超时任务，并封装 CallbackTask 回调任务
  5. 如果自定义的回调函数 InvokeCallback 实现类自己实现了线程池，则使用该线程池来执行 CallbackTask，否则，由当前线程直接执行 CallbackTask。在 CallbackTask 任务中，如果是失败响应，则执行 InvokeCallback#onException 回调函数；如果是成功响应，反序列化响应消息，抽取实际响应信息，然后执行 InvokeCallback#onResponse 回调函数。

二、代码执行流程梯形图

2.1 客户端发出请求

-->RpcClient.invokeWithCallback(String addr, Object request, InvokeCallback invokeCallback, int timeoutMillis)
  -->RpcClientRemoting.invokeWithCallback(String addr, Object request, InvokeContext invokeContext, InvokeCallback invokeCallback, int timeoutMillis) // invokeContext = null
    -->Url url = this.addressParser.parse(addr) // 将 addr 转化为 Url
    -->RpcClientRemoting.invokeWithCallback(Url url, Object request, InvokeContext invokeContext, InvokeCallback invokeCallback, 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.invokeWithCallback(Connection conn, RemotingCommand request, InvokeCallback invokeCallback, int timeoutMillis)
        <!-- 
          4.1 创建 RpcInvokeCallbackListener 对象；
          创建InvokeFuture，并将 RpcInvokeCallbackListener 对象和 invokeCallback 回调函数对象设置到 InvokeFuture 中，
          最后将 {invokeId : InvokeFuture实例} 存储到 Connection 的 Map<Integer, InvokeFuture> invokeFutureMap 
        -->
        -->new RpcInvokeCallbackListener(String address) // RpcInvokeCallbackListener#onResponse(InvokeFuture future) 执行了真正的回调函数
        -->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发送消息

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

总结：

  1. 获取连接
  1. 检查连接
  1. 使用 RpcCommandFactory 创建请求对象 + 序列化
  1. 发起请求
  • 创建 RpcInvokeCallbackListener 对象；创建InvokeFuture，并将 RpcInvokeCallbackListener 对象和 invokeCallback 回调函数对象设置到 InvokeFuture 中，最后将将 {invokeId : InvokeFuture实例} 存储到 Connection 的Map<Integer, InvokeFuture> invokeFutureMap
  • 创建 io.netty.util.Timeout 实例并设置到future实例中
  • 使用 netty 发送消息

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

总结：

与 Sync 完全相同

  1. 创建 InvokeContext 和 RemotingContext
  1. 根据 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 填充响应 + 取消 io.netty.util.TimerTask 超时任务 -->
      -->InvokeFuture.putResponse(RemotingCommand response)
        -->this.countDownLatch.countDown() // 解锁等待（实际上对于callback模式此处没有等待）
      -->future.cancelTimeout();
      <!-- 3.3 调用回调 -->
      -->DefaultInvokeFuture.executeInvokeCallback() // 执行相应的 callbackListener
        -->RpcInvokeCallbackListener.onResponse(InvokeFuture future)
          <!-- 3.3.1 创建 CallbackTask 任务对象，并将 InvokeCallback 设置到该对象中；最后执行该任务（如果InvokeCallback对象设置了线程池，则使用该线程池执行，否则，直接使用当前线程执行） -->
          -->InvokeCallback callback = InvokeFuture.getInvokeCallback() // 从InvokeFuture中获取InvokeCallback
          -->CallbackTask task = new CallbackTask(String remoteAddress, InvokeFuture future)
            <!-- CallbackTask.run() -->
            <!-- 获取InvokeCallback + 获取响应消息 -->
            -->InvokeCallback callback = future.getInvokeCallback()
            -->ResponseCommand response = (ResponseCommand) future.waitResponse(0)
            <!-- 如果响应失败：执行异常回调函数 -->
            -->callback.onException(e)
            <!-- 如果响应成功：反序列化响应 + 抽取实际响应信息 + 执行成功回调函数 -->
            -->response.deserialize()
            -->callback.onResponse(rpcResponse.getResponseObject())

总结：

  1. 创建 InvokeContext 和 RemotingContext
  1. 根据 channel 中的附加属性获取相应的 Protocol，之后使用该 Protocol 实例的 CommandHandler 处理消息
  • 从 CommandHandler 中获取 CommandCode 为 RESPONSE 的 RemotingProcessor 实例 RpcResponseProcessor，之后使用 RpcResponseProcessor 进行响应处理
  • 如果RemotingProcessor自定义了线程池executor执行ProcessTask.run()，否则使用ProcessorManager的defaultExecutor
    ProcessTask.run()：
  • 从连接中根据响应 id 获取请求的 InvokeFuture
  • 填充响应 + 取消 io.netty.util.TimerTask 超时任务
  • 执行回调逻辑

创建 CallbackTask 任务对象，并将 InvokeCallback 设置到该对象中；最后执行该任务（如果 InvokeCallback 对象设置了线程池，则使用该线程池执行，否则，直接使用当前线程执行）
CallbackTask.run()

获取InvokeCallback + 获取响应消息，如果是失败响应，则执行 InvokeCallback#onException 回调函数；如果是成功响应，反序列化响应消息，抽取实际响应信息，然后执行 InvokeCallback#onResponse 回调函数。

最后，看一下 回调监听器 RpcInvokeCallbackListener 的轮廓。

public class RpcInvokeCallbackListener implements InvokeCallbackListener {

    ...

    @Override
    public void onResponse(InvokeFuture future) {
        InvokeCallback callback = future.getInvokeCallback();
        CallbackTask task = new CallbackTask(this.getRemoteAddress(), future);
        if (callback.getExecutor() != null) {
            callback.getExecutor().execute(task);  
        } else {
            task.run();
        }
    }

    class CallbackTask implements Runnable {
        InvokeFuture future;
        
        ...
        
        @Override
        public void run() {
            InvokeCallback callback = future.getInvokeCallback();
            ResponseCommand response = (ResponseCommand) future.waitResponse(0);
            if (response == null || response.getResponseStatus() != ResponseStatus.SUCCESS) {
                ...
                callback.onException(e);
            } else {
                ...
                RpcResponseCommand rpcResponse = (RpcResponseCommand) response;
                response.deserialize();
                callback.onResponse(rpcResponse.getResponseObject());
            } // enf of else
        } // end of run
    }
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容