Dubbo 服务消费者和服务提供者建立通信源码浅析

偏执的太偏执、 2023-10-10 17:14 42阅读 0赞

### Dubbo 在服务消费者和服务提供者之间建立通信 ###

当服务消费者（Consumer）调用服务提供者（Provider）的服务时，Dubbo 底层主要通过以下几个步骤来找到并调用相应的类和方法：

1.  服务引用（Reference）：在消费者端，我们使用 `@DubboReference` 注解引用远程服务。Dubbo 通过解析该注解，创建一个代理对象，用于将对服务接口的调用转发到远程服务提供者。这个代理对象是通过 `org.apache.dubbo.rpc.proxy.ProxyFactory` 接口的实现类（默认为 `org.apache.dubbo.rpc.proxy.javassist.JavassistProxyFactory`）创建的。
2.  服务调用（Invocation）：当消费者调用服务接口的方法时，代理对象会将方法名、参数类型和参数值封装成一个 `org.apache.dubbo.rpc.Invocation` 对象。这个对象包含了调用的所有信息，包括服务接口、方法名、参数类型和参数值等。
3.  服务导出（Export）：在服务提供者端，我们使用 `@DubboService` 注解暴露服务。Dubbo 会扫描这个注解，并通过 `org.apache.dubbo.config.ServiceConfig` 类将服务导出。在这个过程中，会创建一个 `org.apache.dubbo.rpc.Invoker` 对象，它负责将 `Invocation` 对象转发到具体的服务实现类。
4.  序列化和反序列化：服务调用信息（即 `Invocation` 对象）需要通过网络传输，因此需要进行序列化。Dubbo 支持多种序列化协议，默认使用 Hessian2 序列化。序列化和反序列化过程由 `org.apache.dubbo.remoting.Codec` 接口的实现类（如 `org.apache.dubbo.remoting.transport.codec.TransportCodec`）完成。
5.  服务调用过程：当服务提供者接收到来自消费者的请求后，Dubbo 会先反序列化请求数据，然后通过 `org.apache.dubbo.rpc.Invoker` 找到具体的服务实现类和方法。这个过程是通过 `org.apache.dubbo.rpc.filter.ClassLoaderFilter` 和 `org.apache.dubbo.rpc.filter.EchoFilter` 完成的。最后，Dubbo 使用 Java 反射机制调用相应的方法，并将结果序列化后返回给消费者。
6.  异常处理：如果在服务调用过程中发生异常，Dubbo 会将异常信息封装成一个 `org.apache.dubbo.rpc.Result` 对象，并通过序列化返回给消费者。消费者在接收到异常信息后，会将其转换为对应的 Java 异常对象，并抛出。

通过以上步骤，Dubbo 能够在服务消费者和服务提供者之间建立通信，实现远程方法调用。

#### 服务引用（Reference） ####

当我们在消费者端使用 `@DubboReference` 注解引用远程服务时，Dubbo 会在启动过程中解析该注解。具体来说，当 Spring 容器完成初始化后，Dubbo 会通过 `org.apache.dubbo.config.spring.beans.factory.annotation.ReferenceAnnotationBeanPostProcessor` 类处理所有带有 `@DubboReference` 注解的字段或方法。

在处理 `@DubboReference` 注解时，`ReferenceAnnotationBeanPostProcessor` 类会调用 `org.apache.dubbo.config.ReferenceConfig` 类创建一个代理对象。这个代理对象是用来将对服务接口的调用转发到远程服务提供者的。

`ReferenceConfig` 类在创建代理对象时，会通过 `org.apache.dubbo.rpc.proxy.ProxyFactory` 接口的实现类来实现。在 Dubbo 中，默认的实现类是 `org.apache.dubbo.rpc.proxy.javassist.JavassistProxyFactory`。以下是 `JavassistProxyFactory` 类中创建代理对象的关键代码：

public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
          
        return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));}

在这段代码中，`Proxy.getProxy(interfaces)` 方法根据服务接口创建一个动态代理类。然后，`newInstance(new InvokerInvocationHandler(invoker))` 方法创建该动态代理类的实例，并将一个 `InvokerInvocationHandler` 实例传递给它。这样，当我们调用代理对象的方法时，实际上会调用 `InvokerInvocationHandler` 的 `invoke` 方法。而 `InvokerInvocationHandler` 的 `invoke` 方法会将请求转发到远程服务提供者。

总之，通过 `@DubboReference` 注解引用远程服务的过程可以归纳为以下步骤：

1.  解析带有 `@DubboReference` 注解的字段或方法。
2.  使用 `ReferenceConfig` 类创建一个代理对象。
3.  通过 `ProxyFactory` 接口（默认实现为 `JavassistProxyFactory`）创建代理类和代理实例。
4.  为代理实例提供一个 `InvokerInvocationHandler`，用于将方法调用转发到远程服务提供者。

接下来，我们详细了解一下`InvokerInvocationHandler`的`invoke`方法的实现和如何将请求转发到远程服务提供者。

`InvokerInvocationHandler` 是 `org.apache.dubbo.rpc.proxy.InvokerInvocationHandler` 类的实例。它的作用是将方法调用委托给一个实现了`org.apache.dubbo.rpc.Invoker` 接口的对象。`InvokerInvocationHandler` 类的关键代码如下：

public class InvokerInvocationHandler implements InvocationHandler {
        
    
        private static final Logger LOGGER = LoggerFactory.getLogger(InvokerInvocationHandler.class);
        private final Invoker<?> invoker;
    
        public InvokerInvocationHandler(Invoker<?> handler) {
        
            this.invoker = handler;
        }
    
        @Override
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        
            String methodName = method.getName();
            Class<?>[] parameterTypes = method.getParameterTypes();
            // 省略一些非核心代码...  
    
            // 创建 Invocation 对象  
            RpcInvocation rpcInvocation = new RpcInvocation(method, invoker.getInterface(), args);
            rpcInvocation.setTargetServiceUniqueName(invoker.getUrl().getServiceKey());
            rpcInvocation.setInvoker(invoker);
            // 调用 Invoker 的 invoke 方法  
            Result result = invoker.invoke(rpcInvocation);
            // 处理返回结果  
            if (result.hasException()) {
        
                Throwable e = result.getException();
                // 省略异常处理相关代码...  
    
                throw e;
            } else {
        
                return result.getValue();
            }
        }
    }

在 `invoke` 方法中，首先创建一个 `RpcInvocation` 对象，将方法名、参数类型和参数值封装进去。然后，调用 `Invoker` 的 `invoke` 方法，将 `RpcInvocation` 对象传递给远程服务提供者。最后，处理远程调用的结果，如果有异常，则抛出；否则，返回结果值。

`Invoker` 接口的实现类负责具体的远程通信。在 Dubbo 中，默认的实现是基于 Netty 的 `org.apache.dubbo.remoting.transport.netty4.NettyClient`。`NettyClient` 通过 Netty 框架与远程服务提供者建立 TCP 连接，发送请求数据并接收响应数据。当远程调用完成后，`InvokerInvocationHandler` 的 `invoke` 方法会返回结果给代理对象，从而使消费者能够像调用本地方法一样调用远程服务。

总结一下，`@DubboReference` 注解引用远程服务的主要过程包括：

1.  创建代理对象，将方法调用委托给 `InvokerInvocationHandler`。
2.  将方法调用信息封装为 `RpcInvocation` 对象。
3.  调用 `Invoker` 的 `invoke` 方法，实现远程通信。
4.  处理远程调用结果，并将结果返回给代理对象。

这样，消费者可以像调用本地方法一样调用远程服务，而底层的远程通信和序列化细节都由 Dubbo 框架负责处理。

#### 服务调用（Invocation） ####

在消费者调用服务接口的方法时，代理对象通过 `InvokerInvocationHandler` 将方法名、参数类型和参数值等信息封装成一个 `org.apache.dubbo.rpc.Invocation` 对象。我们将从 `InvokerInvocationHandler` 的 `invoke` 方法开始，看一下这个过程是如何实现的。

`InvokerInvocationHandler` 类的 `invoke` 方法如下：

public class InvokerInvocationHandler implements InvocationHandler {
        
        private final Invoker<?> invoker;
    
        public InvokerInvocationHandler(Invoker<?> handler) {
        
            this.invoker = handler;
        }
    
        @Override
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                // 省略部分代码...  
    
            // 创建 RpcInvocation 对象  
            RpcInvocation rpcInvocation = new RpcInvocation(method, invoker.getInterface(), args);
            rpcInvocation.setTargetServiceUniqueName(invoker.getUrl().getServiceKey());
            rpcInvocation.setInvoker(invoker);
            // 调用 Invoker 的 invoke 方法  
            Result result = invoker.invoke(rpcInvocation);
            // 处理返回结果  
            // 省略部分代码...  
    
            return result.getValue();
        }
    }

在 `invoke` 方法中，首先创建一个 `RpcInvocation` 对象。`RpcInvocation` 是 `org.apache.dubbo.rpc.Invocation` 接口的一个实现类。创建 `RpcInvocation` 对象时，传递了 `method`（被调用方法）、`invoker.getInterface()`（服务接口）和 `args`（方法参数值）。接下来，将 `invoker.getUrl().getServiceKey()`（目标服务的唯一名称）和 `invoker`（调用者）设置到 `RpcInvocation` 对象中。这样，`RpcInvocation` 对象就包含了调用的所有信息，如服务接口、方法名、参数类型和参数值等。

创建 `RpcInvocation` 对象后，接下来调用 `Invoker` 的 `invoke` 方法，并将 `RpcInvocation` 对象作为参数传递。`Invoker` 接口的实现类负责实际的远程通信，将封装好的调用信息发送到远程服务提供者。在收到服务提供者的响应后，`invoke` 方法会处理返回结果，如异常处理等，最后返回结果值。

综上所述，在消费者调用服务接口的方法时，代理对象通过 `InvokerInvocationHandler` 的 `invoke` 方法将方法名、参数类型和参数值等信息封装成一个 `org.apache.dubbo.rpc.Invocation` 对象（具体为 `RpcInvocation` 对象）。然后，将 `RpcInvocation` 对象传递给 `Invoker` 的 `invoke` 方法，实现远程调用。

#### 服务导出（Export） ####

在服务提供者端，我们使用 `@DubboService` 注解暴露服务。Dubbo 在启动过程中会扫描这个注解。具体来说，当 Spring 容器完成初始化后，Dubbo 会通过 `org.apache.dubbo.config.spring.beans.factory.annotation.ServiceAnnotationBeanPostProcessor` 类处理所有带有 `@DubboService` 注解的类。

在处理 `@DubboService` 注解时，`ServiceAnnotationBeanPostProcessor` 类会创建一个 `org.apache.dubbo.config.ServiceConfig` 对象，并调用其 `export()` 方法来导出服务。以下是 `ServiceConfig` 类中的关键代码片段：

public class ServiceConfig<T> extends AbstractServiceConfig {
        
    
        // 省略部分代码...  
    
        public synchronized void export() {
                // 省略检查和配置相关代码...  
    
            // 根据协议导出服务  
            doExportUrls();
        }
    
        private void doExportUrls() {
                // 省略部分代码...  
    
            for (ProtocolConfig protocolConfig : protocols) {
                    // 根据协议导出服务  
                doExportUrlsFor1Protocol(protocolConfig, registryURLs);
            }
        }
    
        private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
                // 省略部分代码...  
    
            // 创建 Invoker 对象  
            Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, local);        // 创建 Exporter 对象，并将 Invoker 对象加入缓存  
            Exporter<?> exporter = protocol.export(invoker);
            exporters.add(exporter);
            // 省略部分代码...  
        }
    }

在 `ServiceConfig.export()` 方法中，首先调用 `doExportUrls()` 方法根据配置的协议导出服务。在 `doExportUrlsFor1Protocol()` 方法中，创建一个 `Invoker` 对象。这个对象负责将 `Invocation` 对象转发到具体的服务实现类。创建 `Invoker` 对象时，传递了 `ref`（服务实现类实例）、`interfaceClass`（服务接口）和 `local`（本地服务 URL）。

创建 `Invoker` 对象后，将其传递给 `protocol.export()` 方法，创建一个 `Exporter` 对象。`Exporter` 对象负责将服务暴露给远程消费者。同时，将创建的 `Exporter` 对象添加到 `exporters` 集合中。

此过程中，关键的 `proxyFactory.getInvoker()` 方法调用是通过 `org.apache.dubbo.rpc.proxy.ProxyFactory` 接口的实现类完成的。在 Dubbo 中，默认的实现类是 `org.apache.dubbo.rpc.proxy.javassist.JavassistProxyFactory`。创建的 `Invoker` 对象在接收到消费者请求时，负责将 `Invocation` 对象转发到具体的服务实现类。

总结一下，在服务提供者端，`@DubboService` 注解暴露服务的主要过程包括：

1.  扫描带有 `@DubboService` 注解的类。
2.  创建 `ServiceConfig` 对象并调用其 `export()` 方法导出服务。
3.  根据协议配置创建 `Invoker` 对象，并将其传递给 `protocol.export()` 方法创建 `Exporter` 对象。`Invoker` 对象负责将 `Invocation` 对象转发到具体的服务实现类。

#### 序列化和反序列化 ####

1.  序列化和反序列化：服务调用信息（即 `Invocation` 对象）需要通过网络传输，因此需要进行序列化。Dubbo 支持多种序列化协议，默认使用 Hessian2 序列化。序列化和反序列化过程由 `org.apache.dubbo.remoting.Codec` 接口的实现类（如 `org.apache.dubbo.remoting.transport.codec.TransportCodec`）完成。

以下是 `TransportCodec` 类中的关键代码片段：

public class TransportCodec implements Codec2 {
        
    
        // 省略部分代码...  
    
        @Override
        public void encode(Channel channel, ChannelBuffer buffer, Object message) throws IOException {
                // 获取 ExchangeCodec 实例  
            Codec2 codec = getCodec(channel);        // 调用 ExchangeCodec 的 encode 方法进行序列化  
            codec.encode(channel, buffer, message);
        }
    
        @Override
        public Object decode(Channel channel, ChannelBuffer buffer) throws IOException {
                // 获取 ExchangeCodec 实例  
            Codec2 codec = getCodec(channel);        // 调用 ExchangeCodec 的 decode 方法进行反序列化  
            return codec.decode(channel, buffer);
        }
        // 省略部分代码...  
    }

在 `TransportCodec` 类中，`encode()` 方法负责将 `Invocation` 对象序列化为字节流，`decode()` 方法负责将字节流反序列化为 `Invocation` 对象。实际上，`TransportCodec` 类作为一个适配器，将序列化和反序列化的工作委托给了 `ExchangeCodec` 类。`ExchangeCodec` 类根据用户配置的序列化协议，进一步调用相应的序列化实现类（如 `org.apache.dubbo.common.serialize.hessian2.Hessian2Serialization`）来完成序列化和反序列化工作。

结合源码，我们了解了服务调用信息（即 `Invocation` 对象）的序列化和反序列化过程。这个过程由 `org.apache.dubbo.remoting.Codec` 接口的实现类（如 `org.apache.dubbo.remoting.transport.codec.TransportCodec`）完成。实际的序列化和反序列化工作是由 `ExchangeCodec` 类根据用户配置的序列化协议调用相应的序列化实现类来完成的。

#### 服务调用过程 ####

在服务提供者接收到消费者的请求后，Dubbo 首先会进行反序列化操作，以获取请求数据。这个过程由 `org.apache.dubbo.remoting.Codec` 接口的实现类（如 `org.apache.dubbo.remoting.transport.codec.TransportCodec`）完成。具体的反序列化工作是在 `TransportCodec.decode()` 方法中进行的，如之前所述。

接下来，我们将详细讨论服务调用过程中涉及的组件和流程。

1.  `org.apache.dubbo.rpc.Invoker`：`Invoker` 对象是服务提供者端的关键组件，负责将 `Invocation` 对象转发到具体的服务实现类。`Invoker` 是在 `ServiceConfig.export()` 方法中创建的，如之前所述。
2.  `org.apache.dubbo.rpc.filter.ClassLoaderFilter`：这个过滤器主要负责在调用过程中切换类加载器，以保证服务实现类能够被正确加载。以下是 `ClassLoaderFilter` 类中的关键代码片段：

public class ClassLoaderFilter implements Filter {
        
    
        @Override
        public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        
            ClassLoader ocl = Thread.currentThread().getContextClassLoader();
            Thread.currentThread().setContextClassLoader(invoker.getInterface().getClassLoader());
            try {
        
                return invoker.invoke(invocation);
            } finally {
        
                Thread.currentThread().setContextClassLoader(ocl);
            }
        }
    }

`ClassLoaderFilter.invoke()` 方法在调用过程中切换类加载器，然后调用 `invoker.invoke()` 方法将 `Invocation` 对象传递给下一个过滤器或实际的服务实现类。

1.  `org.apache.dubbo.rpc.filter.EchoFilter`：这个过滤器主要用于处理 `$echo` 方法调用。`$echo` 方法是 Dubbo 内置的一个特殊方法，用于测试服务提供者是否可用。以下是 `EchoFilter` 类中的关键代码片段：

public class EchoFilter implements Filter {
        
    
        @Override
        public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
        
            if (inv.getMethodName().equals("$echo")) {
        
                Object[] args = inv.getArguments();
                if (args.length == 1) {
        
                    return new RpcResult(args[0]);
                } else {
        
                    return new RpcResult(inv.getArguments());
                }
            }
            return invoker.invoke(inv);
        }
    }

`EchoFilter.invoke()` 方法检查方法名是否为 `$echo`。如果是 `$echo` 方法调用，直接返回结果；否则，将 `Invocation` 对象传递给下一个过滤器或实际的服务实现类。

1.  Java 反射机制：在 `Invoker` 对象最终传递给服务实现类时，Dubbo 使用 Java 反射机制调用相应的方法。这个过程发生在 `org.apache.dubbo.rpc.proxy.AbstractProxyInvoker` 类中：

public abstract class AbstractProxyInvoker<T> implements Invoker<T> {
        
    
        // 省略部分代码...  
    
        @Override
        public Result invoke(Invocation invocation) throws RpcException {
        
            try {
                    // 调用 doInvoke 方法，该方法是抽象方法，需要子类实现  
                return new RpcResult(doInvoke(invocation));
            } catch (InvocationTargetException e) {
          // 处理反射调用过程中的异常  
                Throwable te = e.getTargetException();
                if (!(te instanceof RuntimeException) && (te instanceof Exception)) {
        
                    return new RpcResult((Exception) te);
                } else {
        
                    throw new RpcException(te instanceof RuntimeException ? te.getMessage() : "", te);
                }
            } catch (RpcException e) {
          // 处理 Dubbo 框架抛出的异常  
                throw e;
            } catch (Throwable e) {
          // 处理其他异常  
                throw new RpcException(e.getMessage(), e);
            }
        }
    
        // 抽象方法，由子类实现  
        protected abstract Object doInvoke(Invocation invocation) throws Throwable;
    
    // 省略部分代码...  
    }

在 `AbstractProxyInvoker.invoke()` 方法中，Dubbo 调用了 `doInvoke()` 方法，该方法是抽象方法，需要子类实现。具体的实现类是 `org.apache.dubbo.rpc.proxy.javassist.JavassistProxyFactory.JavassistInvoker`，它会使用 Java 反射机制调用服务实现类的方法。

1.  结果序列化：在服务提供者处理完消费者的请求后，会将结果序列化后返回给消费者。这个过程与服务调用信息的序列化和反序列化过程相同，也是由 `org.apache.dubbo.remoting.Codec` 接口的实现类（如 `org.apache.dubbo.remoting.transport.codec.TransportCodec`）完成。实际的序列化工作是在 `TransportCodec.encode()` 方法中进行的，如之前所述。

通过以上分析，我们了解了 Dubbo 服务调用过程中的关键组件和流程，包括服务调用信息的反序列化、过滤器处理、Java 反射机制以及结果序列化等。

#### 异常处理 ####

异常处理是 Dubbo 服务调用过程中的一个重要环节。在服务提供者和消费者之间传输过程中，如果发生异常，Dubbo 会对异常信息进行适当的处理。以下是异常处理过程的关键步骤：

1.  服务提供者处理异常：在服务提供者处理请求时，如果发生异常，Dubbo 会在 `org.apache.dubbo.rpc.proxy.AbstractProxyInvoker.invoke()` 方法中捕获异常，如之前所述。根据异常类型，Dubbo 会将异常封装成一个 `RpcResult` 对象。以下是 `AbstractProxyInvoker.invoke()` 方法中的关键代码片段：

@Override
        public Result invoke(Invocation invocation) throws RpcException {
        
            try {
        
                return new RpcResult(doInvoke(invocation));
            } catch (InvocationTargetException e) {
        
                Throwable te = e.getTargetException();
                if (!(te instanceof RuntimeException) && (te instanceof Exception)) {
        
                    return new RpcResult((Exception) te);
                } else {
        
                    throw new RpcException(te instanceof RuntimeException ? te.getMessage() : "", te);
                }
            } catch (RpcException e) {
        
                throw e;
            } catch (Throwable e) {
        
                throw new RpcException(e.getMessage(), e);
            }
        }

1.  服务提供者序列化异常信息：在服务提供者处理完异常后，会将 `RpcResult` 对象序列化，然后通过网络返回给消费者。这个过程与正常结果的序列化过程相同，由 `org.apache.dubbo.remoting.Codec` 接口的实现类（如 `org.apache.dubbo.remoting.transport.codec.TransportCodec`）完成。实际的序列化工作是在 `TransportCodec.encode()` 方法中进行的。
2.  消费者反序列化异常信息：当消费者接收到服务提供者返回的异常信息后，会使用相应的序列化协议对其进行反序列化，以获取 `RpcResult` 对象。这个过程由 `org.apache.dubbo.remoting.Codec` 接口的实现类（如 `org.apache.dubbo.remoting.transport.codec.TransportCodec`）完成。实际的反序列化工作是在 `TransportCodec.decode()` 方法中进行的。
3.  消费者处理异常：在消费者端，如果接收到的 `RpcResult` 对象包含异常信息，Dubbo 会将其转换为对应的 Java 异常对象，并抛出。这个过程发生在 `org.apache.dubbo.rpc.protocol.AbstractInvoker` 类的 `invoke()` 方法中。以下是关键代码片段：

public class AbstractInvoker<T> implements Invoker<T> {
          
        // 省略部分代码...  
    
        @Override    public Result invoke(Invocation inv) throws RpcException {
                // 调用服务提供者  
            Result asyncResult = doInvoke(inv);  
            // 检查 RpcResult 是否包含异常信息  
            if (asyncResult.hasException()) {
                    // 将异常信息转换为对应的 Java 异常对象，并抛出  
                Throwable exception = asyncResult.getException();            if (exception instanceof RuntimeException) {
                        throw (RuntimeException) exception;            } else {
                        throw new RpcException(exception.getMessage(), exception);            }        }  
            // 返回正常结果  
    return asyncResult;  
    }
    
    // 抽象方法，由子类实现  
    protected abstract Result doInvoke(Invocation invocation) throws Throwable;  
    
    // 省略部分代码...  
    }

通过以上分析，我们了解了 Dubbo 在服务调用过程中如何处理异常信息，包括服务提供者处理异常、序列化异常信息、消费者反序列化异常信息以及消费者处理异常等。