性能文章>源码分析Dubbo服务提供者、服务消费者并发度控制机制>

源码分析Dubbo服务提供者、服务消费者并发度控制机制原创

2年前
259945

本文将详细分析<dubbo:service executes=""/>与<dubbo:reference actives = “”/>的实现机制,深入探讨Dubbo自身的保护机制。

源码分析ExecuteLimitFilter


@Activate(group = Constants.PROVIDER, value = Constants.EXECUTES_KEY )
  • 过滤器作用
    服务调用方并发度控制。
  • 使用场景
    对Dubbo服务提供者实现的一种保护机制,控制每个服务的最大并发度。
  • 阻断条件
    当服务调用超过允许的并发度后,直接抛出RpcException异常。
    接下来源码分析ExecuteLimitFilter的实现细节。

ExecuteLimitFilter#invoke

 1public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
 2        URL url = invoker.getUrl();
 3        String methodName = invocation.getMethodName();
 4        Semaphore executesLimit = null;
 5        boolean acquireResult = false;
 6        int max = url.getMethodParameter(methodName, Constants.EXECUTES_KEY, 0);      // @1
 7        if (max > 0) {
 8            RpcStatus count = RpcStatus.getStatus(url, invocation.getMethodName());             // @2
 9            executesLimit = count.getSemaphore(max);                                                              // @3
10            if(executesLimit != null && !(acquireResult = executesLimit.tryAcquire())) {              // @4
11                throw new RpcException("Failed to invoke method " + invocation.getMethodName() + " in provider " + url + ", cause: The service using threads 
12                      greater than <dubbo:service executes=\"" + max + "\" /> limited.");
13            }
14        }
15        boolean isSuccess = true;
16        try {
17            Result result = invoker.invoke(invocation);                 // @5
18            return result;
19        } catch (Throwable t) {
20            isSuccess = false;
21            if (t instanceof RuntimeException) {
22                throw (RuntimeException) t;
23            } else {
24                throw new RpcException("unexpected exception when ExecuteLimitFilter", t);
25            }
26        } finally {
27            if(acquireResult) {                                   // @6
28                executesLimit.release();
29            }
30        }
31    }
  • 代码@1:从服务提供者列表中获取参数executes的值,如果该值小于等于0,表示不启用并发度控制,直接沿着调用链进行调用。
  • 代码@2:根据服务提供者url和服务调用方法名,获取RpcStatus。
 1public static RpcStatus getStatus(URL url, String methodName) {
 2        String uri = url.toIdentityString();      
 3        ConcurrentMap<String, RpcStatus> map = METHOD_STATISTICS.get(uri);         
 4        if (map == null) {
 5            METHOD_STATISTICS.putIfAbsent(uri, new ConcurrentHashMap<String, RpcStatus>());    
 6            map = METHOD_STATISTICS.get(uri);
 7        }
 8        RpcStatus status = map.get(methodName);          /
 9        if (status == null) {
10            map.putIfAbsent(methodName, new RpcStatus());
11            status = map.get(methodName);
12        }
13        return status;
14    }

这里是并发容器ConcurrentHashMap的经典使用,从这里可以看出ConcurrentMap<String, ConcurrentMap< String, RpcStatus>> METHOD_STATISTICS的存储结构为 { 服务提供者URL唯一字符串:{方法名:RpcStatus} }。

  • 代码@3:根据服务提供者配置的最大并发度,创建该服务该方法对应的信号量对象。
 1public Semaphore getSemaphore(int maxThreadNum) {
 2        if(maxThreadNum <= 0) {
 3            return null;
 4        }
 5        if (executesLimit == null || executesPermits != maxThreadNum) {
 6            synchronized (this) {
 7                if (executesLimit == null || executesPermits != maxThreadNum) {
 8                    executesLimit = new Semaphore(maxThreadNum);
 9                    executesPermits = maxThreadNum;
10                }
11            }
12        }
13        return executesLimit;
14    }

使用了双重检测来创建executesLimit 信号量。

  • 代码@4:如果获取不到锁,并不会阻塞等待,而是直接抛出RpcException,服务端的策略是快速抛出异常,供服务调用方(消费者)根据集群策略进行执行,例如重试其他服务提供者。
  • 代码@5:执行真实的服务调用。
  • 代码@6:如果成功申请到信号量,在服务调用结束后,释放信号量。

总结:<dubbo:service executes=""/>的含义是,针对每个服务每个方法的最大并发度。如果超过该值,则直接抛出RpcException。

源码分析ActiveLimitFilter

@Activate(group = Constants.CONSUMER, value = Constants.ACTIVES_KEY )
  • 过滤器作用
    消费端调用服务的并发控制。
  • 使用场景
    控制同一个消费端对服务端某一服务的并发调用度,通常该值应该小于< dubbo:service executes=""/>
  • 阻断条件
    非阻断,但如果超过允许的并发度会阻塞,超过超时时间后将不再调用服务,而是直接抛出超时。

ActiveLimitFilter#invoke

 1public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
 2        URL url = invoker.getUrl();
 3        String methodName = invocation.getMethodName();
 4        int max = invoker.getUrl().getMethodParameter(methodName, Constants.ACTIVES_KEY, 0);    // @1
 5        RpcStatus count = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName());           // @2
 6        if (max > 0) {                                          
 7            long timeout = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.TIMEOUT_KEY, 0);   // @3
 8            long start = System.currentTimeMillis();
 9            long remain = timeout;
10            int active = count.getActive();                                                                                                                                          // @4
11            if (active >= max) {                                                                                                                                                          // @5
12                synchronized (count) {                                                                                                                                                                      
13                    while ((active = count.getActive()) >= max) {                                                                                                     
14                        try {
15                            count.wait(remain);                                                                                                                                      
16                        } catch (InterruptedException e) {
17                        }
18                        long elapsed = System.currentTimeMillis() - start;                               
19                        remain = timeout - elapsed;
20                        if (remain <= 0) {                                                                                                                                             // @6
21                            throw new RpcException("Waiting concurrent invoke timeout in client-side for service:  "
22                                    + invoker.getInterface().getName() + ", method: "
23                                    + invocation.getMethodName() + ", elapsed: " + elapsed
24                                    + ", timeout: " + timeout + ". concurrent invokes: " + active
25                                    + ". max concurrent invoke limit: " + max);
26                        }
27                    }
28                }
29            }
30        }
31        try {
32            long begin = System.currentTimeMillis();
33            RpcStatus.beginCount(url, methodName);        // @7
34            try {
35                Result result = invoker.invoke(invocation);     // @8
36                RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, true);    // @9
37                return result;
38            } catch (RuntimeException t) {
39                RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, false);
40                throw t;
41            }
42        } finally {
43            if (max > 0) {
44                synchronized (count) {
45                    count.notify();     // @10
46                }
47            }
48        }
49    }
  • 代码@1:从Invoker中获取消息端URL中的配置的actives参数,为什么从Invoker中获取的Url是消费端的Url呢?这是因为在消费端根据服务提供者URL创建调用Invoker时,会用服务提供者URL,然后合并消费端的配置属性,其优先级 -D > 消费端 > 服务端。其代码位于:
RegistryDirectory#toInvokers

1URL url = mergeUrl(providerUrl);
  • 代码@2:根据服务提供者URL和调用服务提供者方法,获取RpcStatus。
  • 代码@3:获取接口调用的超时时间,默认为1s。
  • 代码@4:获取当前消费者,针对特定服务,特定方法的并发调用度,active值。
  • 代码@5:如果当前的并发 调用大于等于允许的最大值,则针对该RpcStatus申请锁,并调用其wait(timeout)进行等待,也就是在接口调用超时时间内,还是未被唤醒,则直接抛出超时异常。
  • 代码@6:判断被唤醒的原因是因为等待超时,还是由于调用结束,释放了"名额“,如果是超时唤醒,则直接抛出异常。
  • 代码@7:在一次服务调用前,先将 服务名+方法名对应的RpcStatus的active加一。
  • 代码@8:执行RPC服务调用。
  • 代码@9:记录成功调用或失败调用,并将active减一。
  • 代码@10:最终成功执行,如果开启了actives机制(dubbo:referecnce actives="")时,唤醒等待者。

总结:<dubbo:reference actives=""/> 是控制消费端对单个服务提供者单个服务允许调用的最大并发度。该值的取值不应该大于<dubbo:service executes=""/>的值,并且如果消费者机器的配置,如果性能不尽相同,不建议对该值进行设置。

点赞收藏
分类:标签:
丁威
请先登录,查看4条精彩评论吧
快去登录吧,你将获得
  • 浏览更多精彩评论
  • 和开发者讨论交流,共同进步

为你推荐

Redis stream 用做消息队列完美吗?

Redis stream 用做消息队列完美吗?

Netty源码解析:writeAndFlush

Netty源码解析:writeAndFlush

5
4