dubbo 并发控制源码分析

左手的ㄟ右手 2023-06-27 03:19 4阅读 0赞

1. 问题

> dubbo可以实现接口或方法层面的并发控制，它是如何实现的？

2. 接口方法的并发控制（服务端）

>     <dubbo:service interface="com.foo.BarService" executes="10" />
> 
> 限制该接口的每个方法并发执行不能超过10个

> <dubbo:service interface="com.foo.BarService"> <dubbo:method name="sayHello" executes="10" /> </dubbo:service>

指定接口的某个方法并发限制为10

3.  接口方法针（客户端）

<dubbo:reference interface="com.foo.BarService" actives="10" />

限制 `com.foo.BarService` 的每个方法，每个客户端并发执行（或占用连接的请求数）不能超过 10 个

4. 源码分析

ActiveLimitFilter用于处理actives

ExecuteLimitFilter用于处理executes

>     @Override
>     public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
>         URL url = invoker.getUrl();
>         String methodName = invocation.getMethodName();
>         int max = url.getMethodParameter(methodName, EXECUTES_KEY, 0);  // 获取配置最大并发
>         if (!RpcStatus.beginCount(url, methodName, max)) {
>             throw new RpcException(RpcException.LIMIT_EXCEEDED_EXCEPTION,
>                     "Failed to invoke method " + invocation.getMethodName() + " in provider " +
>                             url + ", cause: The service using threads greater than <dubbo:service executes=\"" + max +
>                             "\" /> limited.");
>         }
>     
>         invocation.setAttachment(EXECUTELIMIT_FILTER_START_TIME, String.valueOf(System.currentTimeMillis()));
>         try {
>             return invoker.invoke(invocation);
>         } catch (Throwable t) {
>             if (t instanceof RuntimeException) {
>                 throw (RuntimeException) t;
>             } else {
>                 throw new RpcException("unexpected exception when ExecuteLimitFilter", t);
>             }
>         }
>     }

>  
> 
>     /**
>      * @param url
>      */
>     public static boolean beginCount(URL url, String methodName, int max) {
>         max = (max <= 0) ? Integer.MAX_VALUE : max;
>         RpcStatus appStatus = getStatus(url); // 从map中获取，不存在则创建
>         RpcStatus methodStatus = getStatus(url, methodName);
>         if (methodStatus.active.incrementAndGet() > max) {  // 并发控制，每次新增，执行结束后，减一
>             methodStatus.active.decrementAndGet();
>             return false;
>         } else {
>             appStatus.active.incrementAndGet();
>             return true;
>         }
>     }

并发控制关键代码：
    methodStatus.active.incrementAndGet() > max

actives源码

区别在于超过并发会等待timeout时长

>     @Override
>     public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
>         URL url = invoker.getUrl();
>         String methodName = invocation.getMethodName();
>         int max = invoker.getUrl().getMethodParameter(methodName, ACTIVES_KEY, 0);
>         final RpcStatus rpcStatus = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName());
>         if (!RpcStatus.beginCount(url, methodName, max)) {
>             long timeout = invoker.getUrl().getMethodParameter(invocation.getMethodName(), TIMEOUT_KEY, 0);
>             long start = System.currentTimeMillis();
>             long remain = timeout;
>             synchronized (rpcStatus) {
>                 while (!RpcStatus.beginCount(url, methodName, max)) {
>                     try {
>                         rpcStatus.wait(remain);
>                     } catch (InterruptedException e) {
>                         // ignore
>                     }
>                     long elapsed = System.currentTimeMillis() - start;
>                     remain = timeout - elapsed;
>                     if (remain <= 0) {
>                         throw new RpcException(RpcException.LIMIT_EXCEEDED_EXCEPTION,
>                                 "Waiting concurrent invoke timeout in client-side for service:  " +
>                                         invoker.getInterface().getName() + ", method: " + invocation.getMethodName() +
>                                         ", elapsed: " + elapsed + ", timeout: " + timeout + ". concurrent invokes: " +
>                                         rpcStatus.getActive() + ". max concurrent invoke limit: " + max);
>                     }
>                 }
>             }
>         }