一、框架设计

在官方Dubbo 开发指南框架设计部分,给出了引用服务时序图:

 另外,在官方Dubbo 用户指南集群容错部分,给出了服务引用的各功能组件关系图:

 本文将根据以上两张图,分析服务引用的实现原理,并进行详细的代码跟踪与解析。

二、原理和源码解析

2.1 创建代理

Dubbo 基于 Spring 的 Schema 扩展实现 XML 配置解析DubboNamespaceHandler 会将 <dubbo:reference> 标签解析为 ReferenceBeanReferenceBean 实现了 FactoryBean,因此当它在代码中有引用时,会调用 ReferenceBean#getObject() 方法进入节点注册和服务发现流程。

ReferenceBean.java

public Object getObject() throws Exception { return get(); 
}

ReferenceConfig.java

public synchronized T get() { if (destroyed){ throw new IllegalStateException("Already destroyed!"); 
    } if (ref == null) { init(); 
    } return ref; 
} 
 
private void init() { //.......忽略 ref = createProxy(map); } 
 
 
private T createProxy(Map<String, String> map) { //.....忽略 invoker = refprotocol.refer(interfaceClass, urls.get(0)); //.....忽略 // 创建服务代理 return (T) proxyFactory.getProxy(invoker); 
}

 2.2 服务发现

因为通过注册中心,因此在 ReferenceConfig.java#createProxy() 方法中,进入 RegistryProtocol.java#refer() 方法。

RegistryProtocol.java

private Cluster cluster; 
 
public void setCluster(Cluster cluster) { this.cluster = cluster; 
} 
 
private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) { 
    RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url); 
    directory.setRegistry(registry); 
    directory.setProtocol(protocol); URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, NetUtils.getLocalHost(), 0, type.getName(), directory.getUrl().getParameters()); if (! Constants.ANY_VALUE.equals(url.getServiceInterface()) && url.getParameter(Constants.REGISTER_KEY, true)) { 
        registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY, 
                Constants.CHECK_KEY, String.valueOf(false))); 
    } directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,  
            Constants.PROVIDERS_CATEGORY  + "," + Constants.CONFIGURATORS_CATEGORY  
            + "," + Constants.ROUTERS_CATEGORY)); return cluster.join(directory); 
}

RegistryDirectory 通过 RegistryDirectory#subscribeUrl() 向 Zookeeper 订阅服务节点信息并 watch 变更,这样就实现了服务自动发现。

2.3 Invoker选取

2.3.1 Cluster

上面我之所以把设置 Cluster 的代码贴上,是因为此处涉及到一个 Dubbo 服务框架核心的概念——微内核和插件机制(此处会单独一篇文章详细介绍):

 

有关 Dubbo 的设计原则,请查看Dubbo《一些设计上的基本常识》。

Cluster 类的定义如下:

Cluster.java

@SPI(FailoverCluster.NAME) 
public interface Cluster { 
 /** 
     * Merge the directory invokers to a virtual invoker. 
     */ 
    @Adaptive <T> Invoker<T> join(Directory<T> directory) throws RpcException; 
}

cluster 的类型是 Cluster$Adaptive,实际上是一个通用的代理类,它会根据 URL 中的 cluster 参数值定位到实际的 Cluster 实现类(默认是 FailoverCluster)。 由于 ExtensionLoader 在实例化对象时,会在实例化完成之后自动套上 Wrapper 类,而 MockerClusterWrapper 就是这样一个 Wrapper。

MockerClusterWrapper.java

public class MockClusterWrapper implements Cluster { 
 private Cluster cluster; 
 public MockClusterWrapper(Cluster cluster) { this.cluster = cluster; 
    } 
 public <T> Invoker<T> join(Directory<T> directory) throws RpcException { return new MockClusterInvoker<T>(directory, this.cluster.join(directory)); 
    } 
}

也就是说,实例化出来的 FailoverCluster 会作为参数赋予 MockerClusterWrapper#cluster,而 MockClusterWrapper 会作为参数赋予 RegistryProtocol#cluster。因此 RegistryProtocol#doRefer() 中调用 cluster.join(directory) 实际上是调用的 MockClusterWrapper#join(directory) 使用这种机制,可以把一些公共的处理放在 Wrapper 类中,实现代码和功能收敛。

MockClusterInvoker.java

public Result invoke(Invocation invocation) throws RpcException { 
    Result result = null; 
     
    String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), 
                             Constants.MOCK_KEY, Boolean.FALSE.toString()).trim();  if (value.length() == 0 || value.equalsIgnoreCase("false")){ //no mock 
        result = this.invoker.invoke(invocation); 
    } else if (value.startsWith("force")) { if (logger.isWarnEnabled()) { 
            logger.info("force-mock: " + invocation.getMethodName() +  
                        " force-mock enabled , url : " +  directory.getUrl()); 
        } //force:direct mock 
        result = doMockInvoke(invocation, null); 
    } else { //fail-mock try { 
            result = this.invoker.invoke(invocation); 
        }catch (RpcException e) { if (e.isBiz()) { throw e; 
            } else { if (logger.isWarnEnabled()) { 
                    logger.info("fail-mock: " + invocation.getMethodName() +  
                            " fail-mock enabled , url : " +  directory.getUrl(), e); 
                } //fail:mock 
                result = doMockInvoke(invocation, e); 
            } 
        } 
    } return result; 
}

这里还涉及到 Dubbo 另外一个核心机制——Mock。Mock 可以在测试中模拟服务调用的各种异常情况,还用来实现服务降级。 MockClusterWrapper.join() 方法可知,实际创建的 ClusterInvoker 是封装了 FailoverClusterInvoker 的 MockerClusterInvoker

MockerClusterInvoker 中,调用之前 Dubbo 会先检查 URL 中是否有 mock 参数(通过服务治理后台 Consumer 端的屏蔽和容错进行设置,或者直接动态设置 mock 参数值),如果存在且以 force 开头,则不发起远程调用直接执行降级逻辑;如果存在且以 fail 开头,则在远程调用异常时才会执行降级逻辑。

因此,通过 MockerClusterWrapper 成功地在 Invoker 中植入了 Mock 机制。

2.3.2 Directory

RegistryProtocol#doRefer() 中可以看到,服务发现过程是通过 RegistryDirectory 向 Zookeeper 订阅来实现的。 先看看 Directory 类之间的关系:

看下 Directory 接口的定义:

Directory.java

public interface Directory<T> extends Node { 
     
    Class<T> getInterface(); 
 
    List<Invoker<T>> list(Invocation invocation) throws RpcException; 
}

Directory 可以看做是对应 Interface 的 Invoker 列表,而这个列表可能是动态变化的,比如注册中心推送变更。

通过 ReferenceConfig#createProxy() 方法可知,StaticDirectory 主要用于多注册中心引用的场景,它的 invoker 列表是通过参数传入的、固定的。在此不做更详细的解析了。

RegistryDirectory 用于使用单注册中心发现服务的场景。RegistryDirectory 没有重写 list() 方法,所以使用 AbstractDirectory#list() 方法:

AbstractDirectory.java

public List<Invoker<T>> list(Invocation invocation) throws RpcException { if (destroyed) { throw new RpcException("Directory already destroyed .url: " + getUrl()); 
    } List<Invoker<T>> invokers = doList(invocation); 
    List<Router> localRouters = this.routers; // local reference if (localRouters != null && !localRouters.isEmpty()) { for (Router router : localRouters) { try { if (router.getUrl() == null ||  
                         router.getUrl().getParameter(Constants.RUNTIME_KEY, false)) { 
                    invokers = router.route(invokers, getConsumerUrl(), invocation); 
                } 
            } catch (Throwable t) { 
                logger.error("Failed to execute router: " + getUrl() +  
                              ", cause: " + t.getMessage(), t); 
            } 
        } 
    } return invokers; 
}

RegistryDirectory.java

/** 
 * 获取 invoker 列表 
 */ public List<Invoker<T>> doList(Invocation invocation) { if (forbidden) { throw new RpcException(RpcException.FORBIDDEN_EXCEPTION, "Forbid consumer " + 
                      NetUtils.getLocalHost() + " access service " + getInterface().getName() +  
                      " from registry " + getUrl().getAddress() + " use dubbo version " + 
                       Version.getVersion() +  
                       ", Please check registry access list (whitelist/blacklist)."); 
    } 
    List<Invoker<T>> invokers = null; 
    Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap;  //本地缓存 if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) { 
        String methodName = RpcUtils.getMethodName(invocation); //根据方法名从本地缓存中获取invoker列表,此处略 //……     } return invokers == null ? new ArrayList<Invoker<T>>(0) : invokers; 
} 
 
 
/** 
 * 节点变更通知 
 */ public synchronized void notify(List<URL> urls) { 
    List<URL> invokerUrls = new ArrayList<URL>(); 
    List<URL> routerUrls = new ArrayList<URL>(); 
    List<URL> configuratorUrls = new ArrayList<URL>(); for (URL url : urls) { 
        String protocol = url.getProtocol(); 
        String category = url.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY); if (Constants.ROUTERS_CATEGORY.equals(category)  || Constants.ROUTE_PROTOCOL.equals(protocol)) { 
            routerUrls.add(url); 
        } else if (Constants.CONFIGURATORS_CATEGORY.equals(category)  || Constants.OVERRIDE_PROTOCOL.equals(protocol)) { 
            configuratorUrls.add(url); 
        } else 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()); 
        } 
    } // configurators  if (configuratorUrls != null && configuratorUrls.size() >0 ){ this.configurators = toConfigurators(configuratorUrls); 
    } // routers if (routerUrls != null && routerUrls.size() >0 ){ 
        List<Router> routers = toRouters(routerUrls); if(routers != null){ // null - do nothing             setRouters(routers); 
        } 
    } 
    List<Configurator> localConfigurators = this.configurators; // local reference // 合并override参数 this.overrideDirectoryUrl = directoryUrl; if (localConfigurators != null && localConfigurators.size() > 0) { for (Configurator configurator : localConfigurators) { this.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl); 
        } 
    } // providers     refreshInvoker(invokerUrls); 
} 
 
 
/** 
 * 根据invokerURL列表转换为invoker列表 
 */ private void refreshInvoker(List<URL> invokerUrls){ //...... Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls) ;// 将URL列表转成Invoker列表 //...... } 
 
 
/** 
 * 合并url参数 顺序为override > -D >Consumer > Provider 
 */ private Map<String, Invoker<T>> toInvokers(List<URL> urls) { //...... URL url = mergeUrl(providerUrl); //...... }

在 dolist() 方法中,如果通过服务治理禁止 Consumer 访问的话,此处直接抛出响应的异常。

RegistryDirectory 实现了 NotifyListener,在 ZK 节点变化时能收到通知更新内存缓存,其中 RegistryDirectory#mergeUrl() 方法中会按照优先级合并参数(动态配置在此处生效)。

服务引用时从内存缓存中获取并返回invoker列表,并根据路由规则再进行一次过滤。

2.3.3 Router

Router 的作用就是从 Directory 的 invoker 列表中刷选出符合路由规则的 invoker 子集。目前 Dubbo 提供了基于IP、应用名和协议等的静态路由功能,功能和实现比较简单,在此不做过多解释。

2.3.4 LoadBalance

通过 Direcotry 和 Router 之后,返回的是可用的 invoker 子集;在发起服务调用时,需要通过 LoadBalance 选择最终的一个目标 invoker。

在上面的 Cluster 章节中我们知道,调用时首先会经过 MockerClusterInvoker 拦截 Mock 设置,然后再根据配置调用实际的 Invoker(默认是 FailoverClusterInvoker)。

FailoverClusterInvoker 继承 AbstractClusterInvoker,在 AbstractClusterInvoker 中:

AbstractClusterInvoker.java

public Result invoke(final Invocation invocation) throws RpcException { 
 
    checkWheatherDestoried(); 
 LoadBalance loadbalance; 
     
    List<Invoker<T>> invokers = list(invocation); if (invokers != null && invokers.size() > 0) { 
        loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class) 
                     .getExtension(invokers.get(0).getUrl() 
                     .getMethodParameter(invocation.getMethodName(), 
                                         Constants.LOADBALANCE_KEY,  
                                         Constants.DEFAULT_LOADBALANCE)); 
    } else { 
        loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class) 
                     .getExtension(Constants.DEFAULT_LOADBALANCE); 
    } 
    RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation); return doInvoke(invocation, invokers, loadbalance); }

FailoverClusterInvoker.java

public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { 
    List<Invoker<T>> copyinvokers = invokers; 
    checkInvokers(copyinvokers, invocation); int len = getUrl().getMethodParameter(invocation.getMethodName(),  
                       Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1; if (len <= 0) { 
        len = 1; 
    } // retry loop. 
    RpcException le = null; // last exception. 
    List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size()); 
    Set<String> providers = new HashSet<String>(len); for (int i = 0; i < len; i++) { //重试时,进行重新选择,避免重试时invoker列表已发生变化. //注意:如果列表发生了变化,那么invoked判断会失效,因为invoker示例已经改变 if (i > 0) { 
            checkWheatherDestoried(); 
            copyinvokers = list(invocation); //重新检查一下             checkInvokers(copyinvokers, invocation); 
        } Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked); 
        invoked.add(invoker); 
        RpcContext.getContext().setInvokers((List)invoked); try { 
            Result result = invoker.invoke(invocation); if (le != null && logger.isWarnEnabled()) { 
                logger.warn("", le); 
            } return result; 
        } catch (RpcException e) { if (e.isBiz()) { // biz exception. throw e; 
            } 
            le = e; 
        } catch (Throwable e) { 
            le = new RpcException(e.getMessage(), e); 
        } finally { 
            providers.add(invoker.getUrl().getAddress()); 
        } 
    } throw new RpcException(); 
}

在调用或重试时,每次都通过 LoadBalance 选出一个 Invoker 进行调用。

至此,调用流程结束。

 

发布评论

分享到:

IT虾米网

微信公众号号:IT虾米 (左侧二维码扫一扫)欢迎添加!

Servlet的工作原理和生命周期详解
你是第一个吃螃蟹的人
发表评论

◎欢迎参与讨论,请在这里发表您的看法、交流您的观点。