相关文章推荐
任性的砖头  ·  【Go】解析X509_x509 ...·  3 年前    · 
瘦瘦的猴子  ·  python - socket ...·  3 年前    · 
眼睛小的西装  ·  立即创建 Azure 免费帐户 | ...·  3 年前    · 
Code  ›  Dubbo源码分析 - 异步改造 | gentryhuang的博客
源码 线程阻塞 try dubbo
https://gentryhuang.com/posts/c812f120/index.html
欢乐的铁链
2 年前
  1. 1. 概述
  2. 2. 调用方式
  3. 3. Dubbo 2.6.x 异步实现
    1. 3.1. 实现方式
    2. 3.2. 存在问题
  4. 4. Dubbo 2.7.x 异步实现
    1. 4.1. 基础模型
      1. 4.1.1. Result
        1. 4.1.1.1. AppResponse
        2. 4.1.1.2. DecodeableRpcResult
          1. 4.1.1.2.1. 属性
          2. 4.1.1.2.2. 解码
        3. 4.1.1.3. CompletableFuture
        4. 4.1.1.4. AsyncRpcResult
          1. 4.1.1.4.1. 属性
          2. 4.1.1.4.2. 获取结果
            1. 4.1.1.4.2.1. 获取 CompletableFuture
            2. 4.1.1.4.2.2. 获取 AppResponse
            3. 4.1.1.4.2.3. 获取结果
            4. 4.1.1.4.2.4. 添加回调
      2. 4.1.2. Invoker
        1. 4.1.2.1. AsyncToSyncInvoker
        2. 4.1.2.2. AbstractProxyInvoker
        3. 4.1.2.3. AbstractInvoker
          1. 4.1.2.3.1. 构造方法
          2. 4.1.2.3.2. invoke 方法
    2. 4.2. Dubbo协议异步实现
      1. 4.2.1. 消费端
      2. 4.2.2. 服务端
        1. 4.2.2.1. 异步实现
          1. 4.2.2.1.1. 定义 CompletableFuture 签名接口
          2. 4.2.2.1.2. 服务消费方
          3. 4.2.2.1.3. 结果
        2. 4.2.2.2. HeaderExchangeHandler 处理请求
        3. 4.2.2.3. DubboProtocol 的处理器
        4. 4.2.2.4. AbstractProxyInvoker
    3. 4.3. HTTP 协议实现
      1. 4.3.1. 消费端
      2. 4.3.2. 服务端
  5. 5. 小结

概述

从 2.7.0 版本开始，Dubbo 的所有异步编程接口开始以 CompletableFuture 为基础，不仅支持了服务提供方的异步执行，而且对当前的异步调用功能进行了增强。异步改造引入了一些功能接口和实现，以及对部分逻辑进行了调整，但底层逻辑并没有改变，本篇文章将对 Dubbo 的异步演进进行介绍。

Dubbo 的远程调用中大致可以分为以上 4 种调用方式：

  • oneway: 客户端发送消息后，不需要接收响应。对于不需要关心服务响应结果的请求适合 oneway 通信。
  • sync: Dubbo 默认的通信方式，即同步调用。
  • async: 异步调用范畴，使用 Future 的方式获取结果。
  • future: 异步调用范畴，使用 CompletableFuture 获取结果，也支持通过 Future 的方式获取结果。

    • 注意： Dubbo 中的调用方式可以分为两大类，oneway 和 twoway ，对于 Dubbo 协议来说，会对这两种方式做分别处理，对于非 Dubbo 协议不会特别区分。

    Dubbo 2.6.x 异步实现

    Dubbo 2.6.x 的异步实现是针对消费端异步，只需指定调用方式为异步调用，并在需要结果的地方从 RpcContext 中取出 Future 获取结果即可。

  • 指定异步调用 
    1
    2
    		 
    <!-- 设置 async = true，表示异步调用。默认是 false，同步调用 -->
    <dubbo:reference id="demoService" check="false" interface="com.alibaba.dubbo.demo.DemoService" async="true"/>
  • 通过上下文取出 Future 
    1
    2
    3
    		 
    String hello = demoService.sayHello("world"); // call remote method
    Future<String> future = RpcContext.getContext().getFuture();
    String result = future.get();

    具体调用实现可以参考 Dubbo异步调用 。

    存在问题

    Dubbo 2.6.x 提供了一定的异步编程能力，但其异步方式存在以下问题：

  • Future 获取方式不够直接，业务方需要从 RpcContext 中获取。如果同时进行多个异步调用，如果使用不当很容易造成上下文污染。
  • Future 接口无法实现自动回调，而且定义的 ResponseFuture (2.7 已经废弃)虽然支持回调，但支持的异步场景有限，并且不支持 Future 间的相互协调。
  • 不支持服务端异步。

    • Dubbo 2.7.x 异步实现

    Dubbo 2.7.x 异步改造是对 Dubbo 2.6.x 异步功能的增强，引入的 CompletableFuture 既支持 Future 又支持 Callback 的调用方式，使用方可以根据需要自行选择。

    基础模型

    Dubbo 2.7.x 对异步实现进行了改造，引入了相关的接口和实现类，异步实现需要这些相关的基础模型配合完成。下面我们先对基础模型进行介绍。

    Result

    Result 相关的继承关系如下图所示：

    在 Dubbo 2.6.x 中统一使用 RpcResult 表示调用结果，Dubbo 2.7.x 中废弃了 RpcResult ，采用以下三个对象表示结果状态。

  • AsyncRpcResult

    表示的是一个异步的、未完成的RPC调用，是在调用链中实际传递的对象。

  • AppResponse

    表示的是服务端返回的具体响应，相当于 Dubbo 2.6.x 中的 RpcResult 。其子类是 DecodeableRpcResult。

  • CompletableFuture

    表示的是服务端返回的结果，由调用端创建，用于封装 AppResponse 对象。其中 DefaultFuture 继承该类。

    三者关系：AppResponse -> CompletableFuture -> AsyncRpcResult ，下面我们对其进行介绍。

    AppResponse 

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13




    

    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    		 
    public class AppResponse implements Result {
        private static final long serialVersionUID = -6925924956850004727L;

        // 响应结果
        private Object result;

        // 返回的异常信息
        private Throwable exception;

        // 返回的附加信息
        private Map<String, Object> attachments = new HashMap<>();

        public AppResponse() {
        }

        public AppResponse(Object result) {
            this.result = result;
        }

        public AppResponse(Throwable exception) {
            this.exception = exception;
        }

        @Override
        public Object recreate() throws Throwable {
            // 1 异常处理
            if (exception != null) {
                // fix issue#619
                try {
                    // get Throwable class
                    Class clazz = exception.getClass();
                    while (!clazz.getName().equals(Throwable.class.getName())) {
                        clazz = clazz.getSuperclass();
                    }
                    // get stackTrace value
                    Field stackTraceField = clazz.getDeclaredField("stackTrace");
                    stackTraceField.setAccessible(true);
                    Object stackTrace = stackTraceField.get(exception);
                    if (stackTrace == null) {
                        exception.setStackTrace(new StackTraceElement[0]);
                    }
                } catch (Exception e) {
                    // ignore
                }
                throw exception;
            }

            // 2 结果
            return result;
        }

        // 省略其它方法 getter/setter 方法
    }

    AppResponse 是调用的实际返回类型，相当于 Dubbo 2.6.x 中的 RpcResult ，理论上不需要实现 Result 接口，这样做是为了兼容。

    DecodeableRpcResult

    属性 
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    		 
    public class DecodeableRpcResult extends AppResponse implements Codec, Decodeable {

        private static final Logger log = LoggerFactory.getLogger(DecodeableRpcResult.class);
        // 通道
        private Channel channel;
        // 序列化类型
        private byte serializationType;
        // 序列化相关的输入流
        private InputStream inputStream;
        // 响应对象
        private Response response;
        // 调用信息
        private Invocation invocation;
        // 标志是否已经解码
        private volatile boolean hasDecoded;

        public DecodeableRpcResult(Channel channel, Response response, InputStream is, Invocation invocation, byte id) {
            Assert.notNull(channel, "channel == null");
            Assert.notNull(response, "response == null");
            Assert.notNull(is, "inputStream == null");
            this.channel = channel;
            this.response = response;
            this.inputStream = is;
            this.invocation = invocation;
            this.serializationType = id;
        }
    }
    解码 
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    		 
    +--- DecodeableRpcResult
        @Override
        public void decode() throws Exception {
            // 没有解码，则进行解码
            if (!hasDecoded && channel != null && inputStream != null) {
                try {
                    // 解码
                    decode(channel, inputStream);
                } catch (Throwable e) {
                    if (log.isWarnEnabled()) {
                        log.warn("Decode rpc result failed: " + e.getMessage(), e);
                    }
                    response.setStatus(Response.CLIENT_ERROR);
                    response.setErrorMessage(StringUtils.toString(e));
                } finally {
                    hasDecoded = true;
                }
            }
        }
    ---
    @Override
        public Object decode(Channel channel, InputStream input) throws IOException {
            if (log.isDebugEnabled()) {
                Thread thread = Thread.currentThread();
                log.debug("Decoding in thread -- [" + thread.getName() + "#" + thread.getId() + "]");
            }

            // 1 确定序列化方式，用于反序列化
            ObjectInput in = CodecSupport.getSerialization(channel.getUrl(), serializationType)
                    .deserialize(channel.getUrl(), input);

            // 2 读取一个 byte 的标志位，其值可能有 6 种
            byte flag = in.readByte();

            // 3 根据标志位判断当前结果中包含的信息，并调用不同的方法进行处理
            switch (flag) {
                case DubboCodec.RESPONSE_NULL_VALUE:
                    break;
                case DubboCodec.RESPONSE_VALUE:
                    handleValue(in);
                    break;
                case DubboCodec.RESPONSE_WITH_EXCEPTION:
                    handleException(in);
                    break;
                case




    
 DubboCodec.RESPONSE_NULL_VALUE_WITH_ATTACHMENTS:
                    handleAttachment(in);
                    break;
                case DubboCodec.RESPONSE_VALUE_WITH_ATTACHMENTS:
                    // 根据 RpcInvocation 中记录的返回值类型读取返回结果，并设置到当前类的 result 字段
                    handleValue(in);
                    // 读取附加信息并设置到当前类的 attachmetns 中
                    handleAttachment(in);
                    break;
                case DubboCodec.RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS:
                    handleException(in);
                    handleAttachment(in);
                    break;
                default:
                    throw new IOException("Unknown result flag, expect '0' '1' '2' '3' '4' '5', but received: " + flag);
            }
            if (in instanceof Cleanable) {
                ((Cleanable) in).cleanup();
            }
            return this;
        }  
    ---
    /**
         * 处理结果
         *
         * @param in
         * @throws IOException
         */
        private void handleValue(ObjectInput in) throws IOException {
            try {
                // 1 返回结果类型
                Type[] returnTypes;
                if (invocation instanceof RpcInvocation) {
                    returnTypes = ((RpcInvocation) invocation).getReturnTypes();
                } else {
                    returnTypes = RpcUtils.getReturnTypes(invocation);
                }
                // 2 根据返回结果类型获取结果
                Object value = null;
                if (ArrayUtils.isEmpty(returnTypes)) {
                    // This almost never happens?
                    value = in.readObject();
                } else if (returnTypes.length == 1) {
                    value = in.readObject((Class<?>) returnTypes[0]);
                } else {
                    value = in.readObject((Class<?>) returnTypes[0], returnTypes[1]);
                }
                // 3 设置结果 result
                setValue(value);
            } catch (ClassNotFoundException e) {
                rethrow(e);
            }
        }

    DecodeableRpcResult 主要对响应结果进行解码，从字节流中获取数据对象。

    CompletableFuture

    CompletableFuture 是 Java 8 提供的异步编程类，Dubbo 2.7.x 中的 DefaultFuture 继承了 CompletableFuture ，Dubbo 协议下对于 twoway 请求都会返回一个 DefaultFuutre 对象。此外， DefaultFuture 支持在请求的时候指定线程池，用来处理请求的响应，具体的我们会在下一篇文章中分析。

    AsyncRpcResult

    AsyncRpcResult 是在调用链中实际传递的对象，表示一个异步的，未完成的RPC调用。注意，它并不是实际的调用结果， AppResponse 才是业务结果。

    属性 
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    		 
    public class AsyncRpcResult implements Result {
        private static final Logger logger = LoggerFactory.getLogger(AsyncRpcResult.class);

        // 当回调发生时，RpcContext 可能已经被更改。即执行 AsyncRpcResult 上添加的回调方法的线程可能先后处理过多个不同的 AsyncRpcResult 。
        // 因此，我们应该保留当前RpcContext实例的引用，并在回调执行之前恢复它。
        private RpcContext storedContext;
        private RpcContext storedServerContext;

        // 此次 RPC 调用关联的线程池。该值用来作为判断是否是 ThreadlessExecutor 行为
        // oneway 一般不会设置该值，twoway 会设置该值。
        private Executor executor;

        // 此次 RPC 调用关联的 Invocation 对象
        private Invocation invocation;

        // 请求返回的对象 （由使用 AsyncRpcResult 方创建）
        private CompletableFuture<AppResponse> responseFuture;

        // 在构造方法中除了接收发送请求返回的 CompletableFuture<AppResponse> 对象，还会保存当前的 RPC 上下文
        public AsyncRpcResult(CompletableFuture<AppResponse> future, Invocation invocation) {
            this.responseFuture = future;
            this.invocation = invocation;
            this.storedContext = RpcContext.getContext();
            this.storedServerContext = RpcContext.getServerContext();
        }
    }

    需要说明的是， responseFuture 属性不仅针对 Dubbo 协议，HTTP等协议调用返回结果也是 CompletableFuture 对象，都是由调用端创建。区别在于 Dubbo 协议一般返回的是 DefaultFuture 对象，而 HTTP 等协议会构造一个 CompletableFuture 对象，我们会在下面内容提到。此外，针对 oneway 调用方式，由于不关心调用结果，因此会创建一个完整状态的 CompletableFuture 对象。而针对 twoway 调用方式，由于关心调用结果，因此在调用的时候会创建 DefaultFuture 对象。

    获取结果

    AsyncRpcResult 获取结果本质上需要先获取发送请求返回的 CompletableFuture ，也就是 responseFuture 属性，然后再从 responseFuture 中获取 AppResponse 对象，最后调用其对应的方法。

    获取 CompletableFuture 
    1
    2
    3
    4
    		 
    +--- AsyncRpcResult
       public CompletableFuture<AppResponse> getResponseFuture() {
            return responseFuture;
        }
    获取 AppResponse 
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    		 
    +--- AsyncRpcResult
        public Result getAppResponse() {
            try {
                // 如果完成，则获取 AppResponse
                if (responseFuture.isDone()) {
                    return responseFuture.get();
                }
            } catch (Exception e) {
                // This should not happen in normal request process;
                logger.error("Got exception when trying to fetch the underlying result from AsyncRpcResult.");
                throw new RpcException(e);
            }
            // 获取默认的 AppResponse
            return createDefaultValue(invocation);
        }
    		 
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    		 
    +--- AsyncRpcResult
       /**
         * 该方法将始终在最大 timeout 等待之后返回：
         * 1. 如果value在超时前返回，则正常返回。
         * 2. 如果timeout之后没有返回值，则抛出TimeoutException。
         *
         * @return
         * @throws InterruptedException
         * @throws ExecutionException
         */
        @Override
        public Result get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
           // 针对 ThreadlessExecutor 的特殊处理，这里调用 waitAndDrain() 等待响应
            if (executor != null && executor instanceof ThreadlessExecutor) {
                ThreadlessExecutor threadlessExecutor = (ThreadlessExecutor) executor;
                threadlessExecutor.waitAndDrain();




    

            }
            return responseFuture.get(timeout, unit);
        }

        @Override
        public Result get() throws InterruptedException, ExecutionException {
            if (executor != null && executor instanceof ThreadlessExecutor) {
                ThreadlessExecutor threadlessExecutor = (ThreadlessExecutor) executor;
                threadlessExecutor.waitAndDrain();
            }
            return responseFuture.get();
        }

    其中 ThreadlessExecutor 是一个特殊的线程池，主要用来解决同步调用模式下的响应，是对线程模型的优化，我们在下一篇文章中进行详细说明。

    获取结果 
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    		 
    +--- AsyncRpcResult
        @Override
        public Object recreate() throws Throwable {
            RpcInvocation rpcInvocation = (RpcInvocation) invocation;
            //  1 如果是服务端的异步实现，则从上下文中取。
            // 为什么？ 因为接口返回的结果是 CompletableFuture,属于异步范畴（服务端的异步），和消费端异步类似。
            if (InvokeMode.FUTURE == rpcInvocation.getInvokeMode()) {
                return RpcContext.getContext().getFuture();
            }
            // 2 获取 AppResponse 中的结果
            return getAppResponse().recreate();
        }

    AsyncRpcResult.recreate() 方法是获取结果的方法，也就是从 AppResponse 中获取结果。

    添加回调

    回调是 Dubbo 2.7.x 异步改造的重要角色， AsyncRpcResult 支持添加回调方法，而这个回调方法会被包装一层并注册到 responseFuture 上，具体实现如下： 

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    		 
    +--- AsyncRpcResult
        public Result whenCompleteWithContext(BiConsumer<Result, Throwable> fn) {
            // 在responseFuture之上注册回调
            this.responseFuture = this.responseFuture.whenComplete((v, t) -> {
                // 将当前线程的 RpcContext 记录到临时属性中，然后将构造函数中存储的 RpcContext 设置到当前线程中，为后面的回调执行做准备
                beforeContext.accept(v, t);
                // 执行回调 （使用的 RpcContext 是回调所属服务方法的调用线程的 RpcContext）
                fn.accept(v, t);
                // 恢复线程原有的 RpcContext
                afterContext.accept(v, t);
            });
            return this;
        }

    在添加回调时，需要使用 beforeContext 和 afterContext 来保证执行回调的线程的 RpcContext 是最初创建 AsyncRpcResult 对象的线程对应的 RpcContext，执行完回调后需要将执行回调的线程的 RpcContext 恢复到原有值。其中 beforeContext 用于保存执行回调线程的 RpcContext，并将最初创建 AsyncRpcResult 对象的线程的 RpcContext 临时设置到执行回调用线程中，为执行回调做准备。 afterContext 用于恢复执行回调用的线程原有的 RpcContext 。具体实现如下： 

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    		 
    +--- AsyncRpcResult
        private RpcContext tmpContext;
        private RpcContext tmpServerContext;

        private BiConsumer<Result, Throwable> beforeContext = (appResponse, t) -> {
            // 将当前线程的 RpcContext 记录到 tmpContext 中
            tmpContext = RpcContext.getContext();
            tmpServerContext = RpcContext.getServerContext();

            // 将构造函数中存储的 RpcContext (也就是创建 AsyncRpcResult 线程的 RpcContext) 设置到当前线程中
            RpcContext.restoreContext(storedContext);
            RpcContext.restoreServerContext(storedServerContext);
        };

        private BiConsumer<Result, Throwable> afterContext = (appResponse, t) -> {
            // 将当前线程的 RpcContext 恢复到原始值
            RpcContext.restoreContext(tmpContext);
            RpcContext.restoreServerContext(tmpServerContext);
        };

    如此一来， AsyncRpcResult 就可以随意添加回调，无需担心 RpcContext 被污染。

    AsyncRpcResult 整个是为异步请求设计的，但是 Dubbo 中默认的请求方式是同步的，那么 Dubbo 又是如何支持同步调用的呢？Dubbo 进行服务引用时，在 AbstractProtocol.refer() 方法中，Dubbo 会将 AbstractProtocol.protocolBindingRefer() 方法实现返回的 Invoker 对象使用 AsyncToSyncInvoker 封装一层，该对象中的调用逻辑会对同步调用专门处理，我们在下面的内容中进行介绍。相比较而言，Dubbo 2.6.x 在 Dubbo 协议做了异步转同步处理，就是在调用时拿到 DefaultFuture 后立即阻塞等待结果。HTTP 协议就没有异步调用支持，而 Dubbo 2.7.x 使用了 AbstractInvoker 对 Future 功能进行统一支持，也就是 HTTP 协议也基本上支持了调用异步。

    Invoker

    AsyncToSyncInvoker

    AsyncToSyncInvoker 描述了异步转同步的逻辑，是对 AsyncRpcResult 获取结果的补充，触发时机是在执行调用的时候。 

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    		 
    public class AsyncToSyncInvoker<T> implements Invoker<T> {
        // 服务引用的 Invoker
        private Invoker<T> invoker;

        public AsyncToSyncInvoker(Invoker<T> invoker) {
            this.invoker = invoker;
        }

        @Override
        public Class<T> getInterface() {
            return invoker.getInterface();
        }

        @Override
        public Result invoke(Invocation invocation) throws RpcException {
            Result asyncResult = invoker.invoke(invocation);
            
            try {
                // 如果是同步调用，则调用 get() 方法，阻塞等待响应返回。
                // 调用的是 AsyncRpcResult.get 方法，其底层调用的是 CompletableFuture 的 get 方法
                if (InvokeMode.SYNC == ((RpcInvocation) invocation).getInvokeMode()) {
                    /**
                     * NOTICE!
                     * must call {@link java.util.concurrent.CompletableFuture#get(long, TimeUnit)} because
                     * {@link java.util.concurrent.CompletableFuture#get()} 被证明有严重的性能下降。
                     */
                    asyncResult.get(Integer.MAX_VALUE, TimeUnit.MILLISECONDS);
                }
            } catch (InterruptedException e) {
                throw new RpcException("Interrupted unexpectedly while waiting for remote result to return!  method: " +
                        invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
            } catch (ExecutionException e) {
                Throwable t = e.getCause();
                if (t instanceof TimeoutException) {
                    throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " +
                            invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
                } else if (t instanceof RemotingException) {
                    throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " +
                            invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
                } else {
                    throw new RpcException(RpcException.UNKNOWN_EXCEPTION, "Fail to invoke remote method: " +
                            invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
                }
            } catch (Throwable e) {
                throw new RpcException(e.getMessage(), e);
            }




    


            // 非同步调用直接返回调用结果 AsyncRpcResult
            return asyncResult;
        }

        // 省略其它代码
    }

    AsyncToSyncInvoker 是 Invoker 的装饰器，负责将异步调用转换成同步调用，也就是调用 AsyncRpcResult 中的 CompletableFuture.get 方法实现同步等待。相比 Dubbo 2.6.x 还是有很大区别的，Dubbo 2.6.x 使用 Future.get 功能阻塞等待，业务线程将处于阻塞等待状态，返回结果时需要消费端 Dubbo 线程池将结果写到 DefaultFuture 中，业务线程才能取出并返回。Dubbo 2.7.x 彻底优化了这种线程模型，关于优化的背景和实现会在下一篇文章中进行介绍，这里先了解即可。

    AbstractProxyInvoker

    AbstractProxyInvoker 是 Dubbo 框架在服务暴露过程中创建的对象，由 ProxyFactory.getInvoker 创建，是对服务接口实现的封装。该过程对 Dubbo 中所有协议一致。

    AbstractInvoker

    Dubbo 在服务引用时会创建消费端的 Invoker ，对于不同的协议创建的 Invoker 有所不同。下面我们对 Dubbo 协议 和 使用 HTTP 通信的协议 创建 Invoker 流程进行分析。其中每个协议创建的 Invoker 都会继承 AbstractInvoker 抽象类，该抽象类中定义了通用的执行逻辑，如调用模式的确定。

    构造方法 
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    		 
    public abstract class AbstractInvoker<T> implements Invoker<T> {

        protected final Logger logger = LoggerFactory.getLogger(getClass());

        // 当前 Invoker 对象封装的业务接口类型
        private final Class<T> type;
        // 当前 Invoker 关联的 URL 对象
        private final URL url;
        // 当前 Invoker 关联的一些附加信息
        private final Map<String, Object> attachment;

        // 标志 Invoker 的状态
        private volatile boolean available = true;
        private AtomicBoolean destroyed = new AtomicBoolean(false);

        // 构造方法
        public AbstractInvoker(Class<T> type, URL url) {
            this(type, url, (Map<String, Object>) null);
        }
        public AbstractInvoker(Class<T> type, URL url, String[] keys) {
            this(type, url, convertAttachment(url, keys));
        }
        public AbstractInvoker(Class<T> type, URL url, Map<String, Object> attachment) {
            if (type == null) {
                throw new IllegalArgumentException("service type == null");
            }
            if (url == null) {
                throw new IllegalArgumentException("service url == null");
            }
            this.type = type;
            this.url = url;
            this.attachment = attachment == null ? null : Collections.unmodifiableMap(attachment);
        }
    }
    invoke 方法 
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    		 
    +--- AbstractInvoker
     @Override
        public Result invoke(Invocation inv) throws RpcException {
            // if invoker is destroyed due to address refresh from registry, let's allow the current invoke to proceed
            if (destroyed.get()) {
                logger.warn("Invoker for service " + this + " on consumer " + NetUtils.getLocalHost() + " is destroyed, "
                        + ", dubbo version is " + Version.getVersion() + ", this invoker should not be used any longer");
            }
            RpcInvocation invocation = (RpcInvocation) inv;
            invocation.setInvoker(this);
            if (CollectionUtils.isNotEmptyMap(attachment)) {
                invocation.addObjectAttachmentsIfAbsent(attachment);
            }

            // 1 从上下文中取出 附加信息
            Map<String, Object> contextAttachments = RpcContext.getContext().getObjectAttachments();
            if (CollectionUtils.isNotEmptyMap(contextAttachments)) {
                invocation.addObjectAttachments(contextAttachments);
            }

            // 2 设置调用模式  SYNC, ASYNC, FUTURE 。注意，oneway 调用方式
            // 根据以下方式确定调用模式：
            // 1) 根据返回值类型是否是 CompletableFuture ，或方法名是 $invokeAsync，则是 FUTURE 模式。这个属于服务端异步。
            // 2) 根据 async 属性，如果设置 async=true ，则是 ASYNC 模式
            // 3) 默认是 SYNC 模式
            invocation.setInvokeMode(RpcUtils.getInvokeMode(url, invocation));
            
            // 3 如果是异步调用的模式，则给本次调用添加一个唯一id (FUTURE 模式不属于)
            RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);

            AsyncRpcResult asyncResult;
            try {

                // 4 调用子类实现的 doInvoker() 方法
                asyncResult = (AsyncRpcResult) doInvoke(invocation);

                // 对调用异常的处理
                // 4.1 创建 CompletableFuture 对象，使用该对象包装 AppResponse 对象
                // 4.2 使用 AppResponse 对象包装异常信息
                // 4.3 使用 AsyncRpcResult 最后包装 CompletableFuture 对象
            } catch (InvocationTargetException e) { // biz exception
                Throwable te = e.getTargetException();
                if (te == null) {
                    asyncResult = AsyncRpcResult.newDefaultAsyncResult(null, e, invocation);
                } else {
                    if (te instanceof RpcException) {
                        ((RpcException) te).setCode(RpcException.BIZ_EXCEPTION);
                    }
                    asyncResult = AsyncRpcResult.newDefaultAsyncResult(null, te, invocation);
                }
            } catch (RpcException e) {
                if (e.isBiz()) {
                    asyncResult = AsyncRpcResult.newDefaultAsyncResult(null, e, invocation);
                } else {
                    throw e;
                }
            } catch (Throwable e) {
                asyncResult = AsyncRpcResult.newDefaultAsyncResult(null, e, invocation);
            }

            // 5 使用 FutureContext 保存 FutureAdapter，FutureAdapter 中会封装 AsyncRpcResult 中的 CompletableFuture 对象
            RpcContext.getContext().setFuture(new FutureAdapter(asyncResult.getResponseFuture()));
            return asyncResult;
        }

    AbstractInvoker 的 invoke 方法是调用服务的模版方法，具体调用细节交给具体子类实现。

  • 设置附加信息到调用信息 Invocation 中。
  • 设置调用模式
    •  
    推荐文章
    任性的砖头  ·  【Go】解析X509_x509 usage=codesigning_芒果黑的博客-CSDN博客
    3 年前
    瘦瘦的猴子  ·  python - socket question: ConnectionResetError: [Errno 54] Connection reset by peer - Stack Overflow
    3 年前
    眼睛小的西装  ·  立即创建 Azure 免费帐户 | Microsoft Azure
    3 年前
    今天看啥   ·   Py中国   ·   codingpro   ·   小百科   ·   link之家   ·   卧龙AI搜索
    删除内容请联系邮箱 2879853325@qq.com
    Code - 代码工具平台
    © 2024 ~ 沪ICP备11025650号