[享学Netflix] 二十四、Hystrix在滑动窗口内统计：BucketedRollingCounterStream、HealthCountsStream

// 泛型上没有确定任何泛型
public abstract class BucketedRollingCounterStream<Event extends HystrixEvent, Bucket, Output> extends BucketedCounterStream<Event, Bucket, Output> {

    private Observable<Output> sourceStream;
    private final AtomicBoolean isSourceCurrentlySubscribed = new AtomicBoolean(false);
	
	// 核心还是在构造函数上
	// 它比父构造器多了一个参数：reduceBucket函数，意思是减少桶的数量
    protected BucketedRollingCounterStream(HystrixEventStream<Event> stream, int numBuckets, int bucketSizeInMs,
                                           Func2<Bucket, Event, Bucket> appendRawEventToBucket,
                                           Func2<Output, Bucket, Output> reduceBucket) {
                                           
        super(stream, numBuckets, bucketSizeInMs, appendRawEventToBucket);
        Func1<Observable<Bucket>, Observable<Output>> reduceWindowToSummary = window -> window.scan(getEmptyOutputValue(), reduceBucket).skip(numBuckets);
        this.sourceStream = bucketedStream      // 数据流，每个对象代表单元窗口产生的桶
                .window(numBuckets, 1)          // 按照滑动窗口桶的个数进行桶的聚集
                .flatMap(reduceWindowToSummary) // 将一系列的桶聚集成最后的数据对象
                .doOnSubscribe(() -> isSourceCurrentlySubscribed.set(true))
                .doOnUnsubscribe(() -> isSourceCurrentlySubscribed.set(false))
                .share()                        // 共享。不同的订阅者看到的数据是一致的
                .onBackpressureDrop();          // 被压流量控制，当消费者消费速度过慢时就丢弃数据，不进行积压
    }

	// 实现父类的方法
    @Override
    public Observable<Output> observe() {
        return sourceStream;
    }
    // 非public方法
	boolean isSourceCurrentlySubscribed() {
        return isSourceCurrentlySubscribed.get();
    }

}

Func1<Observable<Bucket>, Observable<Output>> reduceWindowToSummary = window -> window.scan(getEmptyOutputValue(), reduceBucket).skip(numBuckets);	

// 监听的事件源是HystrixCommandCompletion，所以事件流是HystrixCommandCompletionStream.getInstance(commandKey)
// 一个桶是一个long[]：也就是说一个桶内装有多个指标数据（已经完成对应的时间有N多）
// 输出的也是这些指标数
public class RollingCommandEventCounterStream extends BucketedRollingCounterStream<HystrixCommandCompletion, long[], long[]> {

	// 每个HystrixCommandKey有且仅有一个RollingCommandEventCounterStream 实例
    private static final ConcurrentMap<String, RollingCommandEventCounterStream> streams = new ConcurrentHashMap<String, RollingCommandEventCounterStream>();

	// 它决定了一个空桶应该有多深
	private static final int NUM_EVENT_TYPES = HystrixEventType.values().length;

	// 静态方法，获取实例
	public static RollingCommandEventCounterStream getInstance(HystrixCommandKey commandKey, HystrixCommandProperties properties) { ... }
	public static RollingCommandEventCounterStream getInstance(HystrixCommandKey commandKey, int numBuckets, int bucketSizeInMs) {
		... // 缓存里有直接拿，缓存里木有就new一个
	}
	...
    public static void reset() {
        streams.clear();
    }

    @Override
    long[] getEmptyBucketSummary() {
        return new long[NUM_EVENT_TYPES];
    }
    ...
}

reduceCommandCompletion函数：

		// 最重要的是eventCounts.getCount()方法
		public static final Func2<long[], HystrixCommandCompletion, long[]> appendEventToBucket = (long[] initialCountArray, HystrixCommandCompletion execution) -> {
			// 结果统计：发射数、fallabck数等
			ExecutionResult.EventCounts eventCounts = execution.getEventCounts();
		
			// 遍历所有的事件类型，分别做处理
            for (HystrixEventType eventType: ALL_EVENT_TYPES) {
                switch (eventType) {
                    case EXCEPTION_THROWN: break;
                    default: // 其它事件全部添加到桶里
                        initialCountArray[eventType.ordinal()] += eventCounts.getCount(eventType);
                        break;
                }
            }
            return initialCountArray;
		}

reduceBucket()：

	public static final Func2<long[], long[], long[]> bucketAggregator = (long[] cumulativeEvents, long[] bucketEventCounts) -> {
            for (HystrixEventType eventType: ALL_EVENT_TYPES) {
                switch (eventType) {
                    case EXCEPTION_THROWN:
                        for (HystrixEventType exceptionEventType: HystrixEventType.EXCEPTION_PRODUCING_EVENT_TYPES) {
                            cumulativeEvents[eventType.ordinal()] += bucketEventCounts[exceptionEventType.ordinal()];
                        }
                        break;
                    default:
                        cumulativeEvents[eventType.ordinal()] += bucketEventCounts[eventType.ordinal()];
                        break;
                }
            }
            return cumulativeEvents;
	}

public static class HealthCounts {
    private final long totalCount;
    private final long errorCount;
    private final int errorPercentage;
    ...
    // 增加次数。返回一个新的HealthCounts实例哦~所以它是线程安全的
	public HealthCounts plus(long[] eventTypeCounts) { ... }
	// 空的  属性均是0
	public static HealthCounts empty() { return EMPTY } 
	...
}

reduceCommandCompletion：使用的HystrixCommandMetrics.appendEventToBucket，上面已经介绍
	
reduceBucket函数：
	
	(healthCounts, bucketEventCounts) ->  healthCounts.plus(bucketEventCounts);

    public enum Collapser {
        BATCH_EXECUTED, ADDED_TO_BATCH, RESPONSE_FROM_CACHE;
    }