服务调用-Dubbo篇

拼搏现实的明天。 2022-12-26 08:17 90阅读 0赞

> **关键词：**RPC调用

--------------------

Dubbo源码版本：

<dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>dubbo</artifactId>
        <version>2.6.9</version>
    </dependency>

服务消费者调用：

<dubbo:reference id="animalService" interface="study.rui.dubbo.AnimalService" />
    
    // 接口类型
    @Autowire
    AnimalService animalService;

或

@Reference
    AnimalService animalService

通过上面两种方式，服务消费者就能结合Spring框架使用Dubbo完成RPC的调用了。AnimalService是一个接口，自动注入的是Dubbo生成的接口代理实现类。代理实现类抽象和封装了服务消费者调用的过程。

--------------------

# **█ 接口代理类** #

需要查看服务消费者的调用过程，首先需要查看代理类的源码。使用**Arthas**工具查看代理类源码：

（1）下载Arthas的jar到本地（某个磁盘文件夹）

curl -O https://arthas.aliyun.com/arthas-boot.jar

![20201203201144407.png][]

（2）在arthas-boot.jar所在文件夹下运行：

java -jar arthas-boot.jar

![20201203201224650.png][]

\[1\]、\[2\]...可以是序列号，列表显示的是当前主机上正在运行的Java进程。XMLConsumer是Dubbo消费者进程。

（3）输入XMLConsumer的序列化：3

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70][]

（4）反编译代理对象

jad com.alibaba.dubbo.common.bytecode.proxy0

com.alibaba.dubbo.common.bytecode.proxy0是表示第一个代理对象，以此类推，第二个是proxy1......

（5）代理类内容

/*                                                                           
     * Decompiled with CFR.                                                      
     *                                                                           
     * Could not load the following classes:                                     
     *  study.rui.dubbo.Animal                                                   
     *  study.rui.dubbo.AnimalService                                            
     */                                                                          
    package com.alibaba.dubbo.common.bytecode;                                   
                                                                                 
    import com.alibaba.dubbo.common.URL;                                         
    import com.alibaba.dubbo.common.bytecode.ClassGenerator;                     
    import com.alibaba.dubbo.rpc.service.EchoService;                            
    import java.lang.reflect.InvocationHandler;                                  
    import java.lang.reflect.Method;                                             
    import study.rui.dubbo.Animal;                                               
    import study.rui.dubbo.AnimalService;                                        
                                                                                 
    public class proxy0                                                          
    implements ClassGenerator.DC,                                                
    AnimalService,                                                               
    EchoService {                                                                
        public static Method[] methods;                                          
        private InvocationHandler handler;                                       
                                                                                 
        public void run(Animal animal) {                                         
            Object[] arrobject = new Object[]{animal};                           
            Object object = this.handler.invoke(this, methods[0], arrobject);    
        }                                                                        
                                                                                 
        public void say() {                                                      
            Object[] arrobject = new Object[]{};                                 
            Object object = this.handler.invoke(this, methods[1], arrobject);    
        }                                                                        
                                                                                 
        public void show(URL uRL) {                                              
            Object[] arrobject = new Object[]{uRL};                              
            Object object = this.handler.invoke(this, methods[2], arrobject);    
        }                                                                        
                                                                                 
        @Override                                                                
        public Object $echo(Object object) {                                     
            Object[] arrobject = new Object[]{object};                           
            Object object2 = this.handler.invoke(this, methods[3], arrobject);   
            return object2;                                                      
        }                                                                        
                                                                                 
        public proxy0() {                                                        
        }                                                                        
                                                                                 
        public proxy0(InvocationHandler invocationHandler) {                     
            this.handler = invocationHandler;                                    
        }                                                                        
    }

--------------------

# **█ 服务调用-消费者** #

当服务消费者在代码中通过animalService.say()调用的时候，实际逻辑会被转发到handler中，handler为**InvocationHandler**：

public void say() {                                                      
            Object[] arrobject = new Object[]{};
            // 调用InvocationHandler的invoke方法                                 
            Object object = this.handler.invoke(this, methods[1], arrobject);    
        }

**1、InvocationHandler：**

java.lang.reflect.InvocationHandler是jdk提供的用于动态代理的接口。那在say方法中InvocationHandler的实现类是什么呢，这就需要去查看代理对象生成的时候的代码了。

ReferenceConfig#get() ——> ReferenceConfig#init() ——> ReferenceConfig#createProxy() ——> proxyFactory.getProxy

创建这个代理对象是通过proxyFactory.getProxy(this.invoker);创建的。

（1）proxyFactory.getProxy，最终默认会调用到**JavassistProxyFactory**的getProxy方法：

public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
        // 生成代理对象，handler是InvokerInvocationHandler
        return Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
    }

（2）关于参数invoker，当只有一个注册中心时，this.invoker = refprotocol.refer(this.interfaceClass, (URL)this.urls.get(0)); 参数interfaceClass是服务接口，urls.get(0)获取注册中心URL，即registry://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService?application=demo-consumer......。refprotocol=(Protocol)ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();

@SPI("dubbo")
    public interface Protocol {
        ......
        @Adaptive
        <T> Invoker<T> refer(Class<T> var1, URL var2) throws RpcException;
        ......
    }

refer方法上面标注了@Adaptive注解，通过refprotocol.refer调用方法的时候，会根据URL中属性protocol的值来获取具体的protocol实现类来执行refer方法。this.urls.get(0)中得到的URL中的protocol=registry，所以实际会调用**RegistryProtocol**的refer方法：

public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
        // 将registry://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService
        // 改成 zookeeper://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService
        url = url.setProtocol(url.getParameter("registry", "dubbo")).removeParameter("registry");
        // ZookeeperRegistry
        Registry registry = this.registryFactory.getRegistry(url);
        // type是服务接口类型，不是RegistryService
        if (RegistryService.class.equals(type)) {
            return this.proxyFactory.getInvoker(registry, type, url);
        } else {
            Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded("refer"));
            String group = (String)qs.get("group");
            return group == null || group.length() <= 0 || Constants.COMMA_SPLIT_PATTERN.split(group).length <= 1 && !"*".equals(group) ? this.doRefer(this.cluster, registry, type, url) : this.doRefer(this.getMergeableCluster(), registry, type, url);
        }
    }

this.doRefer(this.cluster, registry, type, url)，参数cluster是通过set方法自动注入的，是自适应Cluster：

// registry = ZookeeperRegistry
    
    private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
        // 创建RegistryDirectory，其实现了NotifyListener接口，能够动态感知注册中心中服务的变化
        RegistryDirectory<T> directory = new RegistryDirectory(type, url);
        // RegistryProtocol
        directory.setRegistry(registry);
        // this.protocol是创建RegistryProtocol的时候，通过set方法自动注入的
        // 是自适应protocol实现类。
        directory.setProtocol(this.protocol);
        ......
        return cluster.join(directory);
    }

Cluster：

@SPI("failover")
    public interface Cluster {
        @Adaptive
        <T> Invoker<T> join(Directory<T> var1) throws RpcException;
    }

join方法上面标注了**@Adaptive**注解，调用cluster.join，会根据参数Directory中的属性URL中的属性cluster的值来获取具体的子类来执行方法逻辑。默认cluster=failover，即**FailoverCluster**。但是cluster的实现类中存在一个包装类：**MockClusterWrapper**。所以通过cluster.join(directory)获取到的是MockClusterWrapper，在MockClusterWrapper中持有FailoverCluster：

public class MockClusterWrapper implements Cluster {
        // cluster = FailoverCluster
        private Cluster cluster;
    
        public MockClusterWrapper(Cluster cluster) {
            this.cluster = cluster;
        }
        // directory = new RegistryDirectory(type, url)
        public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
            // this.cluster.join(directory)就是调用的FailoverCluster，获取到FailoverClusterInvoker
            return new MockClusterInvoker(directory, this.cluster.join(directory));
        }
    }

此时再看say方法中的this.handler.invoke，其中handler是**InvokerInvocationHandler**。InvokerInvocationHandler中持有**MockClusterInvoker**，MockClusterInvoker中持有**FailoverClusterInvoker**。

**2、InvokerInvocationHandler**

功能转发到InvokerInvocationHandler的invoke方法中，对于不是Object对象，也不是toString、hashCode、equals的方法，是调用this.invoker.invoke(new RpcInvocation(method, args)).recreate(); 从上面的源码分析中可以知道，这里的invoker就是MockClusterInvoker：

public class InvokerInvocationHandler implements InvocationHandler {
        // invoker=MockClusterInvoker
        private final Invoker<?> invoker;
        
        // Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker))
        // 通过InvokerInvocationHandler的构造器传参MockClusterWrapper
        public InvokerInvocationHandler(Invoker<?> handler) {
            this.invoker = handler;
        }
    
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
            String methodName = method.getName();
            Class<?>[] parameterTypes = method.getParameterTypes();
            if (method.getDeclaringClass() == Object.class) {
                return method.invoke(this.invoker, args);
                // 对于toString、hashCode、equals这些方法直接调用invoker中的方法
                // 如果invoker中没有重写，就会调用Object中的对应的方法
            } else if ("toString".equals(methodName) && parameterTypes.length == 0) {
                return this.invoker.toString();
            } else if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
                return this.invoker.hashCode();
            } else {
                return "equals".equals(methodName) && parameterTypes.length == 1 ? this.invoker.equals(args[0]) : this.invoker.invoke(new RpcInvocation(method, args)).recreate();
            }
        }
    }

**3、MockClusterInvoker**

this.invoker.invoke(new RpcInvocation(method, args)).recreate(); 调用MockClusterInvoker的invoke方法，**RpcInvocation**封装调用的方法和参数。recreate用于处理和获取最终的调用结果。

// MockClusterInvoker#invoke
    public Result invoke(Invocation invocation) throws RpcException {
        Result result = null;
        // directory是RegistryDirectory，存放着服务提供者的调用URL
        String value = this.directory.getUrl().getMethodParameter(invocation.getMethodName(), "mock", Boolean.FALSE.toString()).trim();
        // mock功能调用
        // mock是调用失败后的处理方法
        if (value.length() != 0 && !value.equalsIgnoreCase("false")) {
            // 强制调用mock方法
            if (value.startsWith("force")) {
                if (logger.isWarnEnabled()) {
                    logger.info("force-mock: " + invocation.getMethodName() + " force-mock enabled , url : " + this.directory.getUrl());
                }
    
                result = this.doMockInvoke(invocation, (RpcException)null);
            } else {
                try {
                    result = this.invoker.invoke(invocation);
                } catch (RpcException var5) {
                    // 调用发生了RpcException异常会进行mock方法容错
                    
                    if (var5.isBiz()) {
                        throw var5;
                    }
    
                    if (logger.isWarnEnabled()) {
                        logger.warn("fail-mock: " + invocation.getMethodName() + " fail-mock enabled , url : " + this.directory.getUrl(), var5);
                    }
    
                    result = this.doMockInvoke(invocation, var5);
                }
            }
        } else {
            // this.invoker是FailoverClusterInvoker
            result = this.invoker.invoke(invocation);
        }
    
        return result;
    }

**4、FailoverClusterInvoker**

MockClusterInvoker中持有的invoker是FailoverClusterInvoker，所以this.invoker.invoke(invocation);调用的是FailoverClusterInvoker中的invoker方法，invoker是继承自AbstractClusterInvoker：

// AbstractClusterInvoker#invoke
    public Result invoke(Invocation invocation) throws RpcException {
        this.checkWhetherDestroyed();
        LoadBalance loadbalance = null;
        // 附件，比如上传文件的调用
        Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
        if (contextAttachments != null && contextAttachments.size() != 0) {
            ((RpcInvocation)invocation).addAttachments(contextAttachments);
        }
        // 获取所有的服务提供者
        List<Invoker<T>> invokers = this.list(invocation);
        // 获取配置的负载均衡算法，默认是RandomLoadBalance随机负载均衡
        if (invokers != null && !invokers.isEmpty()) {
            loadbalance = (LoadBalance)ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(((Invoker)invokers.get(0)).getUrl().getMethodParameter(RpcUtils.getMethodName(invocation), "loadbalance", "random"));
        }
    
        RpcUtils.attachInvocationIdIfAsync(this.getUrl(), invocation);
        // doInvoke抽象方法，由具体的子类实现
        return this.doInvoke(invocation, invokers, loadbalance);
    }

doInvoke是有具体的子类实现的，即调用FailoverClusterInvoker中的doInvoke方法：

public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        ......
        String methodName = RpcUtils.getMethodName(invocation);
        int len = this.getUrl().getMethodParameter(methodName, "retries", 2) + 1;
        ......
        // FailoverClusterInvoker提供了失败重试的集群容错功能，默认失败了会重试2次
        for(int i = 0; i < len; ++i) {
            ......
            // 进行负载均衡，选出一个服务
            Invoker<T> invoker = this.select(loadbalance, invocation, copyinvokers, invoked);
            ......
    
            try {
                // invoker进行调用
                Result result = invoker.invoke(invocation);
                ......
    
                Result var13 = result;
                return var13;
            } 
            ......
        }
    
        throw new RpcException(le != null ? le.getCode() : 0, "Failed to invoke the method " + invocation.getMethodName() + " in the service " + this.getInterface().getName() + ". Tried " + len + " times of the providers " + providers + " (" + providers.size() + "/" + copyinvokers.size() + ") from the registry " + this.directory.getUrl().getAddress() + " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version " + Version.getVersion() + ". Last error is: " + (le != null ? le.getMessage() : ""), (Throwable)(le != null && le.getCause() != null ? le.getCause() : le));
    }

invoker.invoke进行服务调用，invoker是什么？是从List<Invoker<T>> invokers服务列表中经过负载均衡算法选择出来的。在**AbstractClusterInvoker**的invoke方法中通过this.list(invocation)获得的服务提供者列表：

List<Invoker<T>> invokers = this.list(invocation);

protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
        List<Invoker<T>> invokers = this.directory.list(invocation);
        return invokers;
    }

从上面的代码可见，invokers服务列表是从directory中获取到的，这里的directory是**RegistryDirectory**。查看RegistryDirectory的list方法：

**5、RegistryDirectory**

list方法是继承自**AbstractDirectory**：

// AbstractDirectory#list
    public List<Invoker<T>> list(Invocation invocation) throws RpcException {
        if (this.destroyed) {
            throw new RpcException("Directory already destroyed .url: " + this.getUrl());
        } else {
            // doList是抽象方法，右具体的子类实现
            List<Invoker<T>> invokers = this.doList(invocation);
            List<Router> localRouters = this.routers;
            // 服务路由功能
            if (localRouters != null && !localRouters.isEmpty()) {
                Iterator var4 = localRouters.iterator();
    
                while(var4.hasNext()) {
                    Router router = (Router)var4.next();
    
                    try {
                        if (router.getUrl() == null || router.getUrl().getParameter("runtime", false)) {
                            invokers = router.route(invokers, this.getConsumerUrl(), invocation);
                        }
                    } catch (Throwable var7) {
                        logger.error("Failed to execute router: " + this.getUrl() + ", cause: " + var7.getMessage(), var7);
                    }
                }
            }
    
            return invokers;
        }
    }

doList是抽象方法，具体的逻辑由子类实现，即调用RegistryDirectory的doList方法：

public List<Invoker<T>> doList(Invocation invocation) {
        if (this.forbidden) {
            throw new RpcException(4, "No provider available from registry " + this.getUrl().getAddress() + " for service " + this.getConsumerUrl().getServiceKey() + " on consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", please check status of providers(disabled, not registered or in blacklist).");
        } else {
            List<Invoker<T>> invokers = null;
            // methodInvokerMap中缓存着接口方法与其对应的服务提供者列表
            Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap;
            if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) {
                String methodName = RpcUtils.getMethodName(invocation);
                Object[] args = RpcUtils.getArguments(invocation);
                // 先根据方法名+第一个参数值获取
                if (args != null && args.length > 0 && args[0] != null && (args[0] instanceof String || args[0].getClass().isEnum())) {
                    // localMethodInvokerMap中的key是接口的方法名与第一个参数值
                    invokers = (List)localMethodInvokerMap.get(methodName + "." + args[0]);
                }
                // 根据方法名获取
                if (invokers == null) {
                    invokers = (List)localMethodInvokerMap.get(methodName);
                }
                // 获取*
                if (invokers == null) {
                    invokers = (List)localMethodInvokerMap.get("*");
                }
                // 获取localMethodInvokerMap中最后的invokers
                if (invokers == null) {
                    Iterator<List<Invoker<T>>> iterator = localMethodInvokerMap.values().iterator();
                    if (iterator.hasNext()) {
                        invokers = (List)iterator.next();
                    }
                }
            }
    
            return (List)(invokers == null ? new ArrayList(0) : invokers);
        }
    }

doList方法就是去**methodInvokerMap**去获取服务列表。methodInvokerMap是什么时候初始化的呢？应该是在RegistryDirectory创建的时候，RegistryDirectory何时创建的呢，是在ReferenceConfig调用get方法的时候。回顾《服务注册与发现》中可以知道，invokers列表中的每一个invoker是通过this.invoker = refprotocol.refer(this.interfaceClass, (URL)this.urls.get(0));即**RegistryProtocol**的refer方法，在refer中又会调用doRefer方法。最终通过调用链，会调用RegistryDirectory的notify方法：

public synchronized void notify(List<URL> urls) {
        List<URL> invokerUrls = new ArrayList<URL>();
        ...... 
        for (URL url : urls) {
            ......
            // providers，服务提供者
            if (Constants.PROVIDERS_CATEGORY.equals(category)) {
                invokerUrls.add(url);
            } else {
                logger.warn("Unsupported category " + category + " in notified url: " + url + " from registry " + getUrl().getAddress() + " to consumer " + NetUtils.getLocalHost());
            }
        }
        ......
        // providers
        refreshInvoker(invokerUrls);
    }

refreshInvoker：

private void refreshInvoker(List<URL> invokerUrls) {
        ......
            Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls);
            Map<String, List<Invoker<T>>> newMethodInvokerMap = toMethodInvokers(newUrlInvokerMap);
            // 在这里完成了methodInvokerMap的初始化
            this.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap;
            this.urlInvokerMap = newUrlInvokerMap;
        ......
    }

toInvokers：

toInvokers方法完成了将URL转换成Invoker。

private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
        ......
        invoker = new InvokerDelegate<T>(protocol.refer(serviceType, url), url, providerUrl);
        ......
    }

protocol是在创建RegistryDirectory的时候，调用setProtocol设置的。这里的protocol是使用的RegistryProtocol中的protocol，是自适应Protocol。当调用protocol.refer的方法的时候，会根据参数url中属性protocol的值来决定具体的实现类，此时url中protocol的值是dubbo，则调用DubboProtocol：

public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
        optimizeSerialization(url);
        // getClients获取数据传输客户端类
        DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
        invokers.add(invoker);
        // 返回DubboInvoker
        return invoker;
    }

**6、InvokerDelegate**

回到步骤4中的invoker.invoke，经过步骤5，我们知道，最终的invoker是new InvokerDelegate<T>(DubboInvoker, url, providerUrl);所以继续调用InvokerDelegate的invoke方法。invoke是继承了InvokerWrapper的方法：

// InvokerDelegate的invoke方法是继承自InvokerWrapper
    public Result invoke(Invocation invocation) throws RpcException {
        // 此时的invoker是DubboInvoker，是通过构造器传参设置的。
        return invoker.invoke(invocation);
    }

**7、DubboInvoker**

调用DubboInvoker的invoke方法，invoke方法继承自**AbstractInvoker**，在invoke方法中调用抽象方法doInvoke，由具体的子类实现，即调用DubboInvoker的doInvoke方法：

protected Result doInvoke(final Invocation invocation) throws Throwable {
        RpcInvocation inv = (RpcInvocation) invocation;
        final String methodName = RpcUtils.getMethodName(invocation);
        inv.setAttachment(Constants.PATH_KEY, getUrl().getPath());
        inv.setAttachment(Constants.VERSION_KEY, version);
        
        // clients数组是在创建DubboInvoker时，通过getClients获取到的构造器参数
        // ExchangeClient 数据传输客户端
        ExchangeClient currentClient;
        if (clients.length == 1) { 
            currentClient = clients[0];
        } else {
            currentClient = clients[index.getAndIncrement() % clients.length];
        }
        try {
            // 是否异步
            boolean isAsync = RpcUtils.isAsync(getUrl(), invocation);
            // 是否是不需要返回结果的调用，比如发送心跳
            boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
            int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
            if (isOneway) {
                boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
                currentClient.send(inv, isSent);
                RpcContext.getContext().setFuture(null);
                return new RpcResult();
            } else if (isAsync) {
                // 异步调用
                ResponseFuture future = currentClient.request(inv, timeout);
                RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
                return new RpcResult();
            } else {
                // 一般的方法调用
                RpcContext.getContext().setFuture(null);
                return (Result) currentClient.request(inv, timeout).get();
            }
        } catch (TimeoutException e) {
            throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
        } catch (RemotingException e) {
            throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

**8、ExchangeClient**

调用ExchangeClient的request方法。在DubboProtocol中通过getClient方法获取到ExchangeClient 数组。ExchangeClient有三个实现类：**HeaderExchangeClient**（默认的）、**LazyConnectExchangeClient**（懒连接，先会创建对象，实际调用的时候才建立连接）、**ReferenceCountExchangeClient**（共享连接）。不管是LazyConnectExchangeClient，还是ReferenceCountExchangeClient，在其中都会持有HeaderExchangeClient。功能都会转交给HeaderExchangeClient。

以ReferenceCountExchangeClient为例，调用currentClient.request会先调用ReferenceCountExchangeClient的request方法：

public ResponseFuture request(Object request) throws RemotingException {
        // client=HeaderExchangeClient
        return client.request(request);
    }

client.request：

即调用HeaderExchangeClient的request方法：

public ResponseFuture request(Object request) throws RemotingException {
        // channel=HeaderExchangeChannel，是在HeaderExchangeClient构造函数中初始化的：this.channel = new HeaderExchangeChannel(client);
        return channel.request(request);
    }

channel.request：

即调用HeaderExchangeChannel的request方法：

public ResponseFuture request(Object request, int timeout) throws RemotingException {
        ......
            // channel默认是是NettyClient
            channel.send(req);
        ......
    }

channel.send：

即调用**NettyClient**的send方法，send方法是继承自**AbstractClient**：

public void send(Object message, boolean sent) throws RemotingException {
        if (send_reconnect && !isConnected()) {
            connect();
        }
        // 获取到NettyChannel
        Channel channel = getChannel();
        //TODO Can the value returned by getChannel() be null? need improvement.
        if (channel == null || !channel.isConnected()) {
            throw new RemotingException(this, "message can not send, because channel is closed . url:" + getUrl());
        }
        // 调用NettyChannel的send方法
        channel.send(message, sent);
    }

channel.send：

即调用NettyChannel的send方法：

public void send(Object message, boolean sent) throws RemotingException {
        // channel是Netty中的一个类，后面就是通过Netty传输数据了
        ChannelFuture future = channel.write(message);
    }

--------------------

# **█ 服务调用-提供者** #

在服务提供者服务暴露的时候，调用了**RegistryProtocol**的export方法。在export方法中有：

public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
        // 开启ServerSocket
        final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
        ......
    }

doLocalExport：

private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker) {
        // 调用了protocol.export
        final Invoker<?> invokerDelegete = new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker));
        exporter = new ExporterChangeableWrapper<T>((Exporter<T>) protocol.export(invokerDelegete), originInvoker);
    }

protocol是在创建RegistryProtocol，通过set方法自动注入的自适应扩展类。调用export时，会根据URL中的属性protocol的值来调用具体的实现类。经过了getProviderUrl方法处理后，URL中的protocol变成了dubbo，所以会调用DubboProtocol的export方法：

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
        String key = serviceKey(url);
        // 创建DubboExporter
        DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
        // 缓存到map中
        exporterMap.put(key, exporter);
        // 开启Socket服务
        openServer(url);
        ......
    }

openServer：

private void openServer(URL url) {
        // find server.
        String key = url.getAddress();
        //client can export a service which's only for server to invoke
        boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true);
        if (isServer) {
            // 缓存
            ExchangeServer server = serverMap.get(key);
            if (server == null) {
                // 通过createServer创建服务
                serverMap.put(key, createServer(url));
            } else {
                // server supports reset, use together with override
                server.reset(url);
            }
        }
    }

createServer：

private ExchangeServer createServer(URL url) {
        ......
        ExchangeServer server = Exchangers.bind(url, requestHandler);
        ......
    }

当有请求到来时，会触发调用requestHandler的逻辑。

private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {
    
        @Override
        public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
            if (message instanceof Invocation) {
                Invocation inv = (Invocation) message;
                // 获取invoker，就是从缓存的exporterMap中获取
                Invoker<?> invoker = getInvoker(channel, inv);
                ......
                // 调用invoke
                return invoker.invoke(inv);
            }
            ......
                    }
    
            ......
    };

invoker是在**ServiceConfig**的doExportUrlsFor1Protocol方法中通过下面代码创建的：

Invoker<?> invokerHH = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
    DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invokerHH, this);
    
    Exporter<?> exporter = protocol.export(wrapperInvoker);

所以invoker是**DelegateProviderMetaDataInvoker：**

@Override
    public Result invoke(Invocation invocation) throws RpcException {
        return invoker.invoke(invocation);
    }

调用DelegateProviderMetaDataInvoker的invoke方法就是调用了invokerHH的invoke方法。所以要知道invokerHH的内容，invokerHH默认是调用**JavassistProxyFactory**的getInvoker方法：

public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
        // TODO Wrapper cannot handle this scenario correctly: the classname contains '$'
        final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
        return new AbstractProxyInvoker<T>(proxy, type, url) {
            @Override
            protected Object doInvoke(T proxy, String methodName,
                                      Class<?>[] parameterTypes,
                                      Object[] arguments) throws Throwable {
                return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
            }
        };
    }

wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);就是利用反射调用方法，比如：

method.invoke(object, args);

--------------------

通过上面的分析，服务调用过程大致为（黑色框为服务消费者调用链，蓝色框为服务提供者调用链）：

按照箭头方向进行RPC调用，按箭头反方向返回调用结果。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1][]

*具体详细正规的服务调用介绍，请戳Dubbo官网《[http://dubbo.apache.org/zh/docs/v2.7/dev/source/service-invoking-process/][http_dubbo.apache.org_zh_docs_v2.7_dev_source_service-invoking-process]》*

[20201203201144407.png]: /images/20221120/fe9191fe00824f27a3e713c6a16602d5.png
[20201203201224650.png]: /images/20221120/c19b5a355fce4895a0775e0a05c62fe8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70]: /images/20221120/6f7a3dc5bea347c383b17b990396eee8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1]: /images/20221120/9c14b49adf9249eb8edff881f2a66209.png
[http_dubbo.apache.org_zh_docs_v2.7_dev_source_service-invoking-process]: http://dubbo.apache.org/zh/docs/v2.7/dev/source/service-invoking-process/