開篇
?Hystrix的超時(shí)檢測本質(zhì)上通過啟動(dòng)單獨(dú)線程去檢測的，線程的啟動(dòng)的時(shí)間剛好就是任務(wù)超時(shí)的時(shí)間，本質(zhì)上就是這么個(gè)簡單的邏輯。
?Hystrix超時(shí)后會(huì)拋出一個(gè)HystrixTimeoutException的異常。
超時(shí)檢測邏輯
?Hystrix的超時(shí)包括注冊過程和執(zhí)行過程兩個(gè)，注冊過程如下：
執(zhí)行l(wèi)ift(new HystrixObservableTimeoutOperator<R>(_cmd))關(guān)聯(lián)超時(shí)檢測任務(wù)。
在HystrixObservableTimeoutOperator類中，new TimerListener()負(fù)責(zé)創(chuàng)建檢測任務(wù)，HystrixTimer.getInstance().addTimerListener(listener)負(fù)責(zé)關(guān)聯(lián)定時(shí)任務(wù)。
在HystrixObservableTimeoutOperator類中，addTimerListener通過java的定時(shí)任務(wù)服務(wù)scheduleAtFixedRate在延遲超時(shí)時(shí)間后執(zhí)行。
?Hystrix的超時(shí)執(zhí)行過程如下：
在超時(shí)后執(zhí)行 listener.tick()方法后執(zhí)行類TimerListener的tick方法
在TimerListener類的tick方法中執(zhí)行timeoutRunnable.run()后執(zhí)行HystrixContextRunnable的run方法
在HystrixContextRunnable類run方法中執(zhí)行child.onError(new HystrixTimeoutException())實(shí)現(xiàn)超時(shí)。
executeCommandWithSpecifiedIsolation(_cmd).lift(new HystrixObservableTimeoutOperator<R>(_cmd));
private static class HystrixObservableTimeoutOperator<R> implements Operator<R, R> {
final AbstractCommand<R> originalCommand;
public HystrixObservableTimeoutOperator(final AbstractCommand<R> originalCommand) {
this.originalCommand = originalCommand;
}
@Override
public Subscriber<? super R> call(final Subscriber<? super R> child) {
final CompositeSubscription s = new CompositeSubscription();
// if the child unsubscribes we unsubscribe our parent as well
child.add(s);
//capture the HystrixRequestContext upfront so that we can use it in the timeout thread later
final HystrixRequestContext hystrixRequestContext = HystrixRequestContext.getContextForCurrentThread();
TimerListener listener = new TimerListener() {
@Override
public void tick() {
if (originalCommand.isCommandTimedOut.compareAndSet(TimedOutStatus.NOT_EXECUTED, TimedOutStatus.TIMED_OUT)) {
// report timeout failure
originalCommand.eventNotifier.markEvent(HystrixEventType.TIMEOUT, originalCommand.commandKey);
// shut down the original request
s.unsubscribe();
final HystrixContextRunnable timeoutRunnable = new HystrixContextRunnable(
originalCommand.concurrencyStrategy, hystrixRequestContext, new Runnable() {
@Override
public void run() {
child.onError(new HystrixTimeoutException());
}
});
timeoutRunnable.run();
}
}
@Override
public int getIntervalTimeInMilliseconds() {
return originalCommand.properties.executionTimeoutInMilliseconds().get();
}
};
final Reference<TimerListener> tl = HystrixTimer.getInstance().addTimerListener(listener);
// set externally so execute/queue can see this
originalCommand.timeoutTimer.set(tl);
/**
* If this subscriber receives values it means the parent succeeded/completed
*/
Subscriber<R> parent = new Subscriber<R>() {
@Override
public void onCompleted() {
if (isNotTimedOut()) {
// stop timer and pass notification through
tl.clear();
child.onCompleted();
}
}
@Override
public void onError(Throwable e) {
if (isNotTimedOut()) {
// stop timer and pass notification through
tl.clear();
child.onError(e);
}
}
@Override
public void onNext(R v) {
if (isNotTimedOut()) {
child.onNext(v);
}
}
private boolean isNotTimedOut() {
// if already marked COMPLETED (by onNext) or succeeds in setting to COMPLETED
return originalCommand.isCommandTimedOut.get() == TimedOutStatus.COMPLETED ||
originalCommand.isCommandTimedOut.compareAndSet(TimedOutStatus.NOT_EXECUTED,
TimedOutStatus.COMPLETED);
}
};
// if s is unsubscribed we want to unsubscribe the parent
s.add(parent);
return parent;
}
}
public Reference<TimerListener> addTimerListener(final TimerListener listener) {
startThreadIfNeeded();
// add the listener
Runnable r = new Runnable() {
@Override
public void run() {
try {
listener.tick();
} catch (Exception e) {
logger.error("Failed while ticking TimerListener", e);
}
}
};
//這里直接簡單粗暴的scheduleAtFixedRate以超時(shí)時(shí)間作為周期去判斷是否執(zhí)行完成
ScheduledFuture<?> f = executor.get().getThreadPool().scheduleAtFixedRate(r,
listener.getIntervalTimeInMilliseconds(), listener.getIntervalTimeInMilliseconds(), TimeUnit.MILLISECONDS);
return new TimerReference(listener, f);
}
public class HystrixContextRunnable implements Runnable {
private final Callable<Void> actual;
private final HystrixRequestContext parentThreadState;
public HystrixContextRunnable(Runnable actual) {
this(HystrixPlugins.getInstance().getConcurrencyStrategy(), actual);
}
public HystrixContextRunnable(HystrixConcurrencyStrategy concurrencyStrategy, final Runnable actual) {
this(concurrencyStrategy, HystrixRequestContext.getContextForCurrentThread(), actual);
}
public HystrixContextRunnable(final HystrixConcurrencyStrategy concurrencyStrategy,
final HystrixRequestContext hystrixRequestContext, final Runnable actual) {
this.actual = concurrencyStrategy.wrapCallable(new Callable<Void>() {
@Override
public Void call() throws Exception {
actual.run();
return null;
}
});
this.parentThreadState = hystrixRequestContext;
}
@Override
public void run() {
HystrixRequestContext existingState = HystrixRequestContext.getContextForCurrentThread();
try {
// set the state of this thread to that of its parent
HystrixRequestContext.setContextOnCurrentThread(parentThreadState);
// execute actual Callable with the state of the parent
try {
actual.call();
} catch (Exception e) {
throw new RuntimeException(e);
}
} finally {
// restore this thread back to its original state
HystrixRequestContext.setContextOnCurrentThread(existingState);
}
}
}
