Dubbo 代码解析

时间 2019-11-19

标签 dubbo 代码解析栏目 Dubbo 繁體版

原文原文链接

spring集成

在dubbo-config下面的dubbo-config-spring模块下面，有一个类DubboNamespaceHandler，它实现了Spring提供的接口NamespaceHandlerSupport；在该模块的META-INF文件夹下有两个文件： spring.handlers和spring.schemas，这两个文件里面制定了dubbo的namespace的XSD文件的位置以及dubbo的namespace由DubboNamespaceHandler来处理解析。java

这些解析的config类最终都继承自 AbstractConfig。node

public class DubboNamespaceHandler extends NamespaceHandlerSupport {

  static {
      Version.checkDuplicate(DubboNamespaceHandler.class);
  }

  public void init() {
    registerBeanDefinitionParser("application", new DubboBeanDefinitionParser(ApplicationConfig.class, true));
    registerBeanDefinitionParser("module", new DubboBeanDefinitionParser(ModuleConfig.class, true));
    registerBeanDefinitionParser("registry", new DubboBeanDefinitionParser(RegistryConfig.class, true));
    registerBeanDefinitionParser("monitor", new DubboBeanDefinitionParser(MonitorConfig.class, true));
    registerBeanDefinitionParser("provider", new DubboBeanDefinitionParser(ProviderConfig.class, true));
    registerBeanDefinitionParser("consumer", new DubboBeanDefinitionParser(ConsumerConfig.class, true));
    registerBeanDefinitionParser("protocol", new DubboBeanDefinitionParser(ProtocolConfig.class, true));
    registerBeanDefinitionParser("service", new DubboBeanDefinitionParser(ServiceBean.class, true));
    registerBeanDefinitionParser("reference", new DubboBeanDefinitionParser(ReferenceBean.class, false));
    registerBeanDefinitionParser("annotation", new DubboBeanDefinitionParser(AnnotationBean.class, true));
  }

}

rpc协议： dubbo、hessian、http、redis、memcached、webservice、thrift、rest 等
web

ZookeeperRegistry
doRegister 建立地址节点-> doSubscribe 发布 -> 父类FailbackRegistry 的 notify 通告
redis

LoadBalance

@SPI(RandomLoadBalance.NAME)
public interface LoadBalance {
	/**
	 * select one invoker in list.
	 * 
	 * @param invokers invokers.
	 * @param url refer url
	 * @param invocation invocation.
	 * @return selected invoker.
	 */
    @Adaptive("loadbalance")
	<T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException;

}

通常在实际项目配置权重或负载均衡时不在代码中写死，使用默认配置。须要调节时经过dubbo管控台上进行配置算法

<dubbo:service interface="..." loadbalance="roundrobin"/>

<dubbo:reference interface="..." loadbalance="roundrobin"/>

dubbo:service interface="...">
    <dubbo:methodname="..."loadbalance="roundrobin"/>
</dubbo:service>

<dubbo:reference interface="...">
    <dubbo:method name="..."loadbalance="roundrobin"/>
</dubbo:reference>

Random

随机，按权重设置随机几率。 默认策略
一个截面上碰撞的几率高，但调用量越大分布越均匀，并且按几率使用权重后也比较均匀，有利于动态调整提供者权重。spring

获取全部invokers的个数。
遍历全部Invokers, 获取计算每一个invokers的权重，并把权重累计加起来。每相邻的两个invoker比较他们的权重是否同样，有一个不同说明权重不均等
总权重大于零且权重不均等的状况下按总权重获取随机数offset = random.netx(totalWeight);遍历invokers肯定随机数offset落在哪一个片断(invoker上)
权重相同或者总权重为0，根据invokers个数均等选择

RoundRobin

轮循，按公约后的权重设置轮循比率。数组

存在慢的提供者累积请求问题，好比：第二台机器很慢，但没挂，当请求调到第二台时就卡在那，长此以往，全部请求都卡在调到第二台上。
解决办法 : 结合权重，把第二台机(性能低的)的权重设置低一点app

获取轮询key 服务名+方法名；获取可供调用的invokers个数length；设置最大权重的默认值maxWeight=0；设置最小权重的默认值minWeight=Integer.MAX_VALUE；
遍历全部Inokers，比较出得出maxWeight和minWeight
若是权重是不同的：根据key获取自增序列。自增序列加一与最大权重取模默认获得currentWeigth；遍历全部invokers筛选出大于currentWeight的invokers。
设置可供调用的invokers的个数length
自增序列加一并与length取模，从invokers获取invoker

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
        int length = invokers.size();
        int maxWeight = 0;
        int minWeight = Integer.MAX_VALUE;
        for (int i = 0; i < length; i++) {
            int weight = getWeight(invokers.get(i), invocation);
            maxWeight = Math.max(maxWeight, weight);
            minWeight = Math.min(minWeight, weight);
        }
        if (maxWeight > 0 && minWeight < maxWeight) { 
            AtomicPositiveInteger weightSequence = weightSequences.get(key);
            if (weightSequence == null) {
                weightSequences.putIfAbsent(key, new AtomicPositiveInteger());
                weightSequence = weightSequences.get(key);
            }
            int currentWeight = weightSequence.getAndIncrement() % maxWeight;
            List<Invoker<T>> weightInvokers = new ArrayList<Invoker<T>>();
            for (Invoker<T> invoker : invokers) { 
                if (getWeight(invoker, invocation) > currentWeight) {
                    weightInvokers.add(invoker);
                }
            }
            int weightLength = weightInvokers.size();
            if (weightLength == 1) {
                return weightInvokers.get(0);
            } else if (weightLength > 1) {
                invokers = weightInvokers;
                length = invokers.size();
            }
        }
        AtomicPositiveInteger sequence = sequences.get(key);
        if (sequence == null) {
            sequences.putIfAbsent(key, new AtomicPositiveInteger());
            sequence = sequences.get(key);
        }
        return invokers.get(sequence.getAndIncrement() % length);
    }

LeastActive

最少活跃调用数，相同活跃数的随机，活跃数指调用先后计数差。负载均衡

使慢的提供者收到更少请求，由于越慢的提供者的调用先后计数差会越大dom

获取可调用invoker的总个数；初始化最小活跃数，相同最小活跃的个数；相同最小活跃数的下标数组等等。
遍历全部invokers，获取每一个invoker的获取数active和权重。找出最小权重的invoker；若是有相同最小权重的inovkers，将下标记录到数组leastIndexs[]数组中；累计全部的权重到totalWeight变量。
若是invokers的权重不相等且totalWeight大于0，按总权重随机offsetWeight = random.nextInt(totalWeight)，计算随机值在哪一个片断上并返回invoker。
若是invokers的权重相等或者totalWeight等于0，均等随机

活跃计数的功能消费者是在ActiveLimitFilter中设置的:

ConsistentHash

一致性Hash，相同参数的请求老是发到同一提供者。
当某一台提供者挂时，本来发往该提供者的请求，基于虚拟节点，平摊到其它提供者，不会引发剧烈变更。
算法参见：http://en.wikipedia.org/wiki/Consistent_hashing。
缺省只对第一个参数Hash，若是要修改，请配置<dubbo:parameter key="hash.arguments" value="0,1" />
缺省用160份虚拟节点，若是要修改，请配置<dubbo:parameter key="hash.nodes" value="320" />