扒一扒隔离熔断之Hystrix VS Sentinel

public Observable<R> toObservable() {   ...    final Func0<Observable<R>> applyHystrixSemantics = new Func0<Observable<R>>() {        @Override        public Observable<R> call() {            if (commandState.get().equals(CommandState.UNSUBSCRIBED)) {                return Observable.never();            }            return applyHystrixSemantics(_cmd);//1.关键步骤，命令处理        }    };  ...private Observable<R> applyHystrixSemantics(final AbstractCommand<R> _cmd) {    ...    if (circuitBreaker.attemptExecution()) {//1.【断路器相关处理】，之后HystrixCircuitBreaker中展示        ..        if (executionSemaphore.tryAcquire()) {//2.获取信号量，如果是THREAD线程池策略，【直接返回true】，这里需要注意，不然流程将进行不下去            try {                executionResult = executionResult.setInvocationStartTime(System.currentTimeMillis());                return executeCommandAndObserve(_cmd)//3.核心执行方法                        .doOnError(markExceptionThrown)                        .doOnTerminate(singleSemaphoreRelease)                        .doOnUnsubscribe(singleSemaphoreRelease);            } ...}

public boolean allowRequest() {            if (properties.circuitBreakerForceOpen().get()) {                // properties have asked us to force the circuit open so we will allow NO requests                return false;            }            if (properties.circuitBreakerForceClosed().get()) {                // we still want to allow isOpen() to perform it's calculations so we simulate normal behavior                isOpen();                // properties have asked us to ignore errors so we will ignore the results of isOpen and just allow all traffic through                return true;            }            return !isOpen() || allowSingleTest();        }

TimerListener listener = new TimerListener() {                @Override                public void tick() {                    if (originalCommand.isCommandTimedOut.compareAndSet(TimedOutStatus.NOT_EXECUTED, TimedOutStatus.TIMED_OUT)) {                        // 标记事件，可以认为是开的hook，这里暂忽略                        originalCommand.eventNotifier.markEvent(HystrixEventType.TIMEOUT, originalCommand.commandKey);                        //取消原Obserable的订阅                        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();                }            };

protected BucketedRollingCounterStream(HystrixEventStream<Event> stream, final int numBuckets, int bucketSizeInMs,                                           final Func2<Bucket, Event, Bucket> appendRawEventToBucket,                                           final Func2<Output, Bucket, Output> reduceBucket) {        super(stream, numBuckets, bucketSizeInMs, appendRawEventToBucket);        Func1<Observable<Bucket>, Observable<Output>> reduceWindowToSummary = new Func1<Observable<Bucket>, Observable<Output>>() {            @Override            public Observable<Output> call(Observable<Bucket> window) {                return window.scan(getEmptyOutputValue(), reduceBucket).skip(numBuckets);            }        };        this.sourceStream = bucketedStream      //stream broken up into buckets                .window(numBuckets, 1)          //emit overlapping windows of buckets                .flatMap(reduceWindowToSummary) //convert a window of bucket-summaries into a single summary                .doOnSubscribe(new Action0() {                    @Override                    public void call() {                        isSourceCurrentlySubscribed.set(true);                    }                })                .doOnUnsubscribe(new Action0() {                    @Override                    public void call() {                        isSourceCurrentlySubscribed.set(false);                    }                })                .share()                        //multiple subscribers should get same data                .onBackpressureDrop();          //if there are slow consumers, data should not buffer    }

class ListState {    /*     * 这里的data之所以用AtomicReferenceArray而不是普通数组，是因为data需要     * 在不同的ListState对象中跨线程来引用，需要可见性和并发性的保证。     */    private final AtomicReferenceArray<Bucket> data;    private final int size;    private final int tail;    private final int head;    private ListState(AtomicReferenceArray<Bucket> data, int head, int tail) {        this.head = head;        this.tail = tail;        if (head == 0 && tail == 0) {            size = 0;        } else {            this.size = (tail + dataLength - head) % dataLength;        }        this.data = data;    }}