序
本文主要研究一下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 inconsistent
executionHook.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() {
@Override
public void call() {
if (semaphoreHasBeenReleased.compareAndSet(false, true)) {
executionSemaphore.release();
}
}
};
final Action1<Throwable> markExceptionThrown = new Action1<Throwable>() {
@Override
public 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 setup
executionSemaphorePerCircuit.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 isolation
return 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;
}
@Override
public boolean tryAcquire() {
int currentCount = count.incrementAndGet();
if (currentCount > numberOfPermits.get()) {
count.decrementAndGet();
return false;
} else {
return true;
}
}
@Override
public void release() {
count.decrementAndGet();
}
@Override
public int getNumberOfPermitsUsed() {
return count.get();
}
}
内部是使用AtomicInteger来进行计数，tryAcquire方法，是先增，如果超过限制，则再减
TryableSemaphoreNoOp.DEFAULT
/* package */static class TryableSemaphoreNoOp implements TryableSemaphore {
public static final TryableSemaphore DEFAULT = new TryableSemaphoreNoOp();
@Override
public boolean tryAcquire() {
return true;
}
@Override
public void release() {
}
@Override
public int getNumberOfPermitsUsed() {
return 0;
}
}
TryableSemaphoreNoOp.DEFAULT是TryableSemaphoreNoOp的单例，默认放行
小结
hystrix的ExecutionIsolationStrategy分为SEMAPHORE及THREAD模式，在command的执行前后逻辑，内置了对TryableSemaphore的tryAcquire及release操作。只是在获取TryableSemaphore实现类的时候，针对SEMAPHORE模式才真正根据execution.isolation.semaphore.maxConcurrentRequests属性进行限制，而如果是THREAD模式，则返回一个都放行的TryableSemaphoreNoOp实例。
