Dubbo 并發控制

消費端并發控制

<dubbo:reference id="userService" interface="com.test.UserServiceBo"
        group="dubbo" version="1.0.0" timeout="3000" actives="10"/>

在服務消費方設定接口中每個方法并發請求個數，通過設定actives參數。

<dubbo:reference id="userService" interface="com.test.UserServiceBo"
        group="dubbo" version="1.0.0" timeout="3000">
                <dubbo:method name="sayHello" actives="10" />
</dubbo:reference>

在服務消費方設定接口中的某個方法的并發請求個數，通過設定actives參數。

服務端并發控制

<dubbo:service interface="com.test.UserServiceBo" ref="userService"
            group="dubbo"  version="1.0.0" timeout="3000" executes="10"/>

在服務提供方設定接口中每個方法的并發請求數，通過設定executes參數。

<dubbo:service interface="com.test.UserServiceBo" ref="userService"
            group="dubbo" version="1.0.0" timeout="3000" >
            <dubbo:method name="sayHello" executes="10" />
</dubbo:service>

在服務提供方設定接口中某個方法的并發請求數，通過設定executes參數。

消費端并發限制 - ActiveLimitFilter

@Activate(group = Constants.CONSUMER, value = Constants.ACTIVES_KEY)
public class ActiveLimitFilter implements Filter {

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        URL url = invoker.getUrl();
        String methodName = invocation.getMethodName();
        int max = invoker.getUrl().getMethodParameter(methodName, Constants.ACTIVES_KEY, 0);
        // 擷取方法級别的并發限制的RpcStatus對象
        RpcStatus count = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName());
        if (max > 0) {
            long timeout = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.TIMEOUT_KEY, 0);
            long start = System.currentTimeMillis();
            long remain = timeout;
            int active = count.getActive();
            // 通過synchronized和wait實作用戶端并發限制超過時候需要等待直至逾時。
            if (active >= max) {
                synchronized (count) {
                    while ((active = count.getActive()) >= max) {
                        try {
                            count.wait(remain);
                        } catch (InterruptedException e) {
                        }
                        long elapsed = System.currentTimeMillis() - start;
                        remain = timeout - elapsed;
                        if (remain <= 0) {
                            throw new RpcException("Waiting concurrent invoke timeout in client-side for service:  "
                                    + invoker.getInterface().getName() + ", method: "
                                    + invocation.getMethodName() + ", elapsed: " + elapsed
                                    + ", timeout: " + timeout + ". concurrent invokes: " + active
                                    + ". max concurrent invoke limit: " + max);
                        }
                    }
                }
            }
        }

        try {
            long begin = System.currentTimeMillis();
            // 累加方法級别的并發數
            RpcStatus.beginCount(url, methodName);
            try {
                // 執行方法調用
                Result result = invoker.invoke(invocation);
                // 遞減方法級别的并發數
                RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, true);

                return result;
            } catch (RuntimeException t) {
                // 遞減方法級别的并發數
                RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, false);
                throw t;
            }
        } finally {
            // 針對有并發限制的通過notify進行喚醒
            if (max > 0) {
                synchronized (count) {
                    count.notify();
                }
            }
        }
    }

}

1、首先會去獲得服務消費端每服務每方法最大可并行執行請求數。
2、如果方法設定并發請求數就需要判斷是否超并發數，超過并發數就等待直至逾時。
3、按照累加并發數、執行方法、遞減并發數，最後進行喚醒炒作。
4、消費端設定actives時候會等待直至逾時。

服務端并發限制 - ExecuteLimitFilter

@Activate(group = Constants.PROVIDER, value = Constants.EXECUTES_KEY)
public class ExecuteLimitFilter implements Filter {

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        URL url = invoker.getUrl();
        String methodName = invocation.getMethodName();
        Semaphore executesLimit = null;
        boolean acquireResult = false;
        int max = url.getMethodParameter(methodName, Constants.EXECUTES_KEY, 0);
        if (max > 0) {
            RpcStatus count = RpcStatus.getStatus(url, invocation.getMethodName());
//            if (count.getActive() >= max) {
            /**
             * http://manzhizhen.iteye.com/blog/2386408
             * use semaphore for concurrency control (to limit thread number)
             */
            executesLimit = count.getSemaphore(max);
            // 服務提供方設定并發數量後，如果同時請求數量大于了設定的executes的值，則會抛出異常。
            if(executesLimit != null && !(acquireResult = executesLimit.tryAcquire())) {
                throw new RpcException("Failed to invoke method " + invocation.getMethodName() + " in provider " + url + ", cause: The service using threads greater than <dubbo:service executes=\"" + max + "\" /> limited.");
            }
        }
        long begin = System.currentTimeMillis();
        boolean isSuccess = true;
        RpcStatus.beginCount(url, methodName);
        try {
            Result result = invoker.invoke(invocation);
            return result;
        } catch (Throwable t) {
            isSuccess = false;
            if (t instanceof RuntimeException) {
                throw (RuntimeException) t;
            } else {
                throw new RpcException("unexpected exception when ExecuteLimitFilter", t);
            }
        } finally {
            RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, isSuccess);
            // 如果需要擷取過信号量就進行釋放動作。
            if(acquireResult) {
                executesLimit.release();
            }
        }
    }
}

1、首先會去獲得服務提供者每服務每方法最大可并行執行請求數。
2、如果每服務每方法最大可并行執行請求數大于零，那麼就基于基于服務 URL + 方法次元擷取一個RpcStatus執行個體。
3、通過RpcStatus執行個體擷取一個信号量，若果擷取的這個信号量調用tryAcquire傳回false，則抛出異常。
4、如果沒有抛異常，那麼久調用RpcStatus靜态方法beginCount，給這個 URL + 方法次元開始計數。
5、調用服務。
6、調用結束後計數調用RpcStatus靜态方法endCount，計數結束。
7、釋放信号量。
8、需要注意的是，服務提供方設定并發數量後，如果同時請求數量大于了設定的executes的值，則會抛出異常。

并發限制實作核心 - RpcStatus

public class RpcStatus {
    // service級别的并發限制全局變量
    private static final ConcurrentMap<String, RpcStatus> SERVICE_STATISTICS = new ConcurrentHashMap<String, RpcStatus>();
    // method級别的并發限制全局變量
    private static final ConcurrentMap<String, ConcurrentMap<String, RpcStatus>> METHOD_STATISTICS = new ConcurrentHashMap<String, ConcurrentMap<String, RpcStatus>>();
    private final ConcurrentMap<String, Object> values = new ConcurrentHashMap<String, Object>();
    // 記錄活躍的記錄
    private final AtomicInteger active = new AtomicInteger();
    private final AtomicLong total = new AtomicLong();
    private final AtomicInteger failed = new AtomicInteger();
    private final AtomicLong totalElapsed = new AtomicLong();
    private final AtomicLong failedElapsed = new AtomicLong();
    private final AtomicLong maxElapsed = new AtomicLong();
    private final AtomicLong failedMaxElapsed = new AtomicLong();
    private final AtomicLong succeededMaxElapsed = new AtomicLong();
    // 用于記錄服務提供端的限制
    private volatile Semaphore executesLimit;
    private volatile int executesPermits;

    private RpcStatus() {
    }

    // 擷取service級别的并發限制變量
    public static RpcStatus getStatus(URL url) {
        String uri = url.toIdentityString();
        RpcStatus status = SERVICE_STATISTICS.get(uri);
        if (status == null) {
            SERVICE_STATISTICS.putIfAbsent(uri, new RpcStatus());
            status = SERVICE_STATISTICS.get(uri);
        }
        return status;
    }

    // 擷取method級别的并發限制變量，根據service=>method順序查找
    public static RpcStatus getStatus(URL url, String methodName) {
        String uri = url.toIdentityString();
        ConcurrentMap<String, RpcStatus> map = METHOD_STATISTICS.get(uri);
        if (map == null) {
            METHOD_STATISTICS.putIfAbsent(uri, new ConcurrentHashMap<String, RpcStatus>());
            map = METHOD_STATISTICS.get(uri);
        }
        RpcStatus status = map.get(methodName);
        if (status == null) {
            map.putIfAbsent(methodName, new RpcStatus());
            status = map.get(methodName);
        }
        return status;
    }

    // 累加service 和 method的并發限制變量
    public static void beginCount(URL url, String methodName) {
        beginCount(getStatus(url));
        beginCount(getStatus(url, methodName));
    }
    // 并發限制變量的原子累加
    private static void beginCount(RpcStatus status) {
        status.active.incrementAndGet();
    }
    // 遞減service 和 method的并發限制變量
    public static void endCount(URL url, String methodName, long elapsed, boolean succeeded) {
        endCount(getStatus(url), elapsed, succeeded);
        endCount(getStatus(url, methodName), elapsed, succeeded);
    }

    private static void endCount(RpcStatus status, long elapsed, boolean succeeded) {
        status.active.decrementAndGet();
        status.total.incrementAndGet();
        status.totalElapsed.addAndGet(elapsed);
        if (status.maxElapsed.get() < elapsed) {
            status.maxElapsed.set(elapsed);
        }
        if (succeeded) {
            if (status.succeededMaxElapsed.get() < elapsed) {
                status.succeededMaxElapsed.set(elapsed);
            }
        } else {
            status.failed.incrementAndGet();
            status.failedElapsed.addAndGet(elapsed);
            if (status.failedMaxElapsed.get() < elapsed) {
                status.failedMaxElapsed.set(elapsed);
            }
        }
    }

    public Semaphore getSemaphore(int maxThreadNum) {
        if(maxThreadNum <= 0) {
            return null;
        }

        if (executesLimit == null || executesPermits != maxThreadNum) {
            synchronized (this) {
                if (executesLimit == null || executesPermits != maxThreadNum) {
                    executesLimit = new Semaphore(maxThreadNum);
                    executesPermits = maxThreadNum;
                }
            }
        }

        return executesLimit;
    }
}

service級别的并發限制全局變量 private static final ConcurrentMap SERVICE_STATISTICS。
method級别的并發限制全局變量 private static final ConcurrentMap> METHOD_STATISTICS。
記錄活躍的記錄 private final AtomicInteger active。

Dubbo 并發控制

消費端并發控制

服務端并發控制

消費端并發限制 - ActiveLimitFilter

服務端并發限制 - ExecuteLimitFilter

并發限制實作核心 - RpcStatus

繼續閱讀

初識dubbo整合spring出現的service注入失敗的問題

手繪 SpringCloud微服務 VS DubboRPC服務

Dubbo的叢集容錯

Dubbo的負載均衡、叢集容錯、服務降級等機制詳解1. Dubbo與RPC的關系2. Dubbo的基本使用3. dubbo的REST協定4. dubbo的控制台5. dubbo的服務路由

springboot與dubbo版本比對問題

Dubbo|基礎知識之解析自定義标簽流程

Apache Dubbo 雲原生可觀測性的探索與實踐

RPC 分布式服務架構-Dubbo之負載均衡一、概述二、負載均衡政策三、配置

分布式與微服務系列（二）、分布式RPC架構Apache Dubbo服務一、 Apache Dubbo概述二、服務注冊中心Zookeeper三、Dubbo快速入門四、Dubbo管理控制台

持續內建篇【3】：SonarQube代碼品質管理平台的安裝及使用一、SonarQube代碼品質管理平台的安裝二、SonarQube代碼品質管理平台的配置與使用

Dubbo學習記錄1

Dubbo負載均衡政策Dubbo負載均衡政策

手把手實作RPC架構--簡易版Dubbo構造（三）用戶端動态代理

阿裡巴巴分布式服務架構 Dubbo 團隊成員梁飛專訪

阿裡巴巴分布式服務架構Dubbo介紹引言SOA化服務架構DubboDubbo的主要特點

Dubbo：來自于阿裡巴巴的分布式服務架構 Dubbo：來自于阿裡巴巴的分布式服務架構