聊聊hystrix的execution.isolation.semaphore.maxConcurrentRequests属性

序

本文主要研究一下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();}} 这个方法调用了getExecutionSemaphore来获取TryableSemaphore，执行之前进行tryAcquire，执行结束之后进行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;}}

• 这里针对ExecutionIsolationStrategy进行判断，如果是SEMAPHORE，则根据commandKey获取或新建对应的TryableSemaphore
• 创建的话，使用的是TryableSemaphoreActual，其numberOfPermits参数就是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();} 定义了三个方法，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();}} 内部是使用AtomicInteger来进行计数，tryAcquire方法，是先增，如果超过限制，则再减

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的单例，默认放行

小结

hystrix的ExecutionIsolationStrategy分为SEMAPHORE及THREAD模式，在command的执行前后逻辑，内置了对TryableSemaphore的tryAcquire及release操作。只是在获取TryableSemaphore实现类的时候，针对SEMAPHORE模式才真正根据execution.isolation.semaphore.maxConcurrentRequests属性进行限制，而如果是THREAD模式，则返回一个都放行的TryableSemaphoreNoOp实例。

doc

• execution.isolation.semaphore.maxConcurrentRequests

总结

以上是生活随笔为你收集整理的聊聊hystrix的execution.isolation.semaphore.maxConcurrentRequests属性的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。