Dubbo源代碼分析七：使用executes屬性的一個問題

我們知道，在Dubbo中可以給Provider配置線程池大小來控制系統提供服務的最大並行度，默認是200個，如果我們想配置成500，可以如下配置：

當我們想限制某個dubbo服務使用的最大線程數量時，dubbo提供了executes這一屬性來提供這個功能，比如我們想限制某個介面最大能同時使用線程池中的100個線程，我們可以如下配置：

我們看下dubbo內部executes是如何實現的，這就得移步到ExecuteLimitFilter，我們直接看下它的實現：

@Activate(group = Constants.PROVIDER, value = Constants.EXECUTES_KEY)

public class ExecuteLimitFilter implements Filter {

public Result invoke(Invoker invoker, Invocation invocation) throws RpcException {

URL url = invoker.getUrl;

String methodName = invocation.getMethodName;

int max = url.getMethodParameter(methodName, Constants.EXECUTES_KEY, 0);

// 如果該介面/方法設置了executes並且值大於0

if (max > 0) {

// 取出該介面/方法對應的計數器

RpcStatus count = RpcStatus.getStatus(url, invocation.getMethodName);

// 如果當前使用的線程數量已經大於等於設置的閾值，那麼直接拋出異常

if (count.getActive >= max) {

throw new RpcException("Failed to invoke method " + invocation.getMethodName + " in provider " + url + ", cause: The service using threads greater than limited.");

}

}

long begin = System.currentTimeMillis;

boolean isException = false;

// 計數器+1

RpcStatus.beginCount(url, methodName);

try {

Result result = invoker.invoke(invocation);

return result;

} catch (Throwable t) {

isException = true;

if (t instanceof RuntimeException) {

throw (RuntimeException) t;

} else {

throw new RpcException("unexpected exception when ExecuteLimitFilter", t);

}

} finally {

// 在finally中進行計數器-1

RpcStatus.endCount(url, methodName, System.currentTimeMillis - begin, isException);

看上面的代碼，可以得知基本步驟就是（黃底的部分代碼）：計數器當前值和閾值比較 > 計數器+1 > 計數器-1。這種方式在高並發時會出現靜態條件問題的，比如當前該介面已經使用了99個線程，這是時候有兩個請求同時到達都發現count.getActive是小於max的，於是該介面使用的線程數就有可能達到了101個。

那麼，我們能不能把RpcStatus.beginCount(url, methodName);放到count.getActive >= max的前面去執行？仔細想想後也不行，這樣做的話有可能在高並發時請求被count.getActive >= max卡死，因為大量請求將計數器+1（+1的速度遠大於原有請求執行完將計數器-1的速度），導致一段時間內計數器一直大於閾值但實際上該介面使用的線程數卻是0。

於是，為了將比較和+1做成原子的，我們想到了Semaphore，信號量Semaphore維護了一組許可，用來管理有限的資源，比如這裡的線程數。使用Semaphore的有兩點需要注意的地方，第一個就是說如果使用不當會導致Semaphore中的許可數多於最初設置的值，或者變為負數，也就是說，Semaphore並沒有對非正常使用作出任何保護措施。還有一點是Semaphore沒有方法能直接修改許可數量，但我們可以通過間接方法（比如多release幾次以增加許可數量），但畢竟不優雅。

Semaphore可以解決上述問題，但當需要修改線程數時，我們應該新建一個Semaphore對象，當採用這種替換Semaphore對象來應對許可數的變化時，一定要確保在仍和情況下acquire和release操作應該在一個Semaphore對象上。這也是我們在設計那些控制並發的小工具時需要注意的地方。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！