序

本文主要研究一下hystrix的execution.isolation.semaphore.maxConcurrentRequests屬性

AbstractCommand.applyHystrixSemantics

hystrix-core-1.5.12-sources.jar!/com/netflix/hystrix/AbstractCommand.java

private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) {// mark that we're starting execution on the ExecutionHook// if this hook throws an exception, then a fast-fail occurs with no fallback. No state is left inconsistentexecutionHook.onStart(_cmd);/* determine if we're allowed to execute */if (circuitBreaker.attemptExecution()) {final TryableSemaphore executionSemaphore = getExecutionSemaphore();final AtomicBoolean semaphoreHasBeenReleased = new AtomicBoolean(false);final Action0 singleSemaphoreRelease = new Action0() {@Overridepublic void call() {if (semaphoreHasBeenReleased.compareAndSet(false, true)) {executionSemaphore.release();}}};final Action1<Throwable> markExceptionThrown = new Action1<Throwable>() {@Overridepublic void call(Throwable t) {eventNotifier.markEvent(HystrixEventType.EXCEPTION_THROWN, commandKey);}};if (executionSemaphore.tryAcquire()) {try {/* used to track userThreadExecutionTime */executionResult = executionResult.setInvocationStartTime(System.currentTimeMillis());return executeCommandAndObserve(_cmd).doOnError(markExceptionThrown).doOnTerminate(singleSemaphoreRelease).doOnUnsubscribe(singleSemaphoreRelease);} catch (RuntimeException e) {return Observable.error(e);}} else {return handleSemaphoreRejectionViaFallback();}} else {return handleShortCircuitViaFallback();}} 這個(gè)方法調(diào)用了getExecutionSemaphore來(lái)獲取TryableSemaphore，執(zhí)行之前進(jìn)行tryAcquire，執(zhí)行結(jié)束之后進(jìn)行release

AbstractCommand.getExecutionSemaphore

hystrix-core-1.5.12-sources.jar!/com/netflix/hystrix/AbstractCommand.java

/*** Get the TryableSemaphore this HystrixCommand should use for execution if not running in a separate thread.* * @return TryableSemaphore*/protected TryableSemaphore getExecutionSemaphore() {if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.SEMAPHORE) {if (executionSemaphoreOverride == null) {TryableSemaphore _s = executionSemaphorePerCircuit.get(commandKey.name());if (_s == null) {// we didn't find one cache so setupexecutionSemaphorePerCircuit.putIfAbsent(commandKey.name(), new TryableSemaphoreActual(properties.executionIsolationSemaphoreMaxConcurrentRequests()));// assign whatever got set (this or another thread)return executionSemaphorePerCircuit.get(commandKey.name());} else {return _s;}} else {return executionSemaphoreOverride;}} else {// return NoOp implementation since we're not using SEMAPHORE isolationreturn TryableSemaphoreNoOp.DEFAULT;}}

• 這里針對(duì)ExecutionIsolationStrategy進(jìn)行判斷，如果是SEMAPHORE，則根據(jù)commandKey獲取或新建對(duì)應(yīng)的TryableSemaphore
• 創(chuàng)建的話，使用的是TryableSemaphoreActual，其numberOfPermits參數(shù)就是execution.isolation.semaphore.maxConcurrentRequests的值
• 如果ExecutionIsolationStrategy是THREAD的話，這里TryableSemaphore返回的是TryableSemaphoreNoOp.DEFAULT，也就是不做任何操作，都放行

TryableSemaphore

/* package */static interface TryableSemaphore {/*** Use like this:* <p>* * <pre>* if (s.tryAcquire()) {* try {* // do work that is protected by 's'* } finally {* s.release();* }* }* </pre>* * @return boolean*/public abstract boolean tryAcquire();/*** ONLY call release if tryAcquire returned true.* <p>* * <pre>* if (s.tryAcquire()) {* try {* // do work that is protected by 's'* } finally {* s.release();* }* }* </pre>*/public abstract void release();public abstract int getNumberOfPermitsUsed();} 定義了三個(gè)方法，tryAcquire，release、getNumberOfPermitsUsed

TryableSemaphoreActual

/*** Semaphore that only supports tryAcquire and never blocks and that supports a dynamic permit count.* <p>* Using AtomicInteger increment/decrement instead of java.util.concurrent.Semaphore since we don't need blocking and need a custom implementation to get the dynamic permit count and since* AtomicInteger achieves the same behavior and performance without the more complex implementation of the actual Semaphore class using AbstractQueueSynchronizer.*//* package */static class TryableSemaphoreActual implements TryableSemaphore {protected final HystrixProperty<Integer> numberOfPermits;private final AtomicInteger count = new AtomicInteger(0);public TryableSemaphoreActual(HystrixProperty<Integer> numberOfPermits) {this.numberOfPermits = numberOfPermits;}@Overridepublic boolean tryAcquire() {int currentCount = count.incrementAndGet();if (currentCount > numberOfPermits.get()) {count.decrementAndGet();return false;} else {return true;}}@Overridepublic void release() {count.decrementAndGet();}@Overridepublic int getNumberOfPermitsUsed() {return count.get();}} 內(nèi)部是使用AtomicInteger來(lái)進(jìn)行計(jì)數(shù)，tryAcquire方法，是先增，如果超過(guò)限制，則再減

TryableSemaphoreNoOp.DEFAULT

/* package */static class TryableSemaphoreNoOp implements TryableSemaphore {public static final TryableSemaphore DEFAULT = new TryableSemaphoreNoOp();@Overridepublic boolean tryAcquire() {return true;}@Overridepublic void release() {}@Overridepublic int getNumberOfPermitsUsed() {return 0;}} TryableSemaphoreNoOp.DEFAULT是TryableSemaphoreNoOp的單例，默認(rèn)放行

小結(jié)

hystrix的ExecutionIsolationStrategy分為SEMAPHORE及THREAD模式，在command的執(zhí)行前后邏輯，內(nèi)置了對(duì)TryableSemaphore的tryAcquire及release操作。只是在獲取TryableSemaphore實(shí)現(xiàn)類的時(shí)候，針對(duì)SEMAPHORE模式才真正根據(jù)execution.isolation.semaphore.maxConcurrentRequests屬性進(jìn)行限制，而如果是THREAD模式，則返回一個(gè)都放行的TryableSemaphoreNoOp實(shí)例。

doc

• execution.isolation.semaphore.maxConcurrentRequests

總結(jié)

以上是生活随笔為你收集整理的聊聊hystrix的execution.isolation.semaphore.maxConcurrentRequests属性的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。