[享学Netflix] 二十二、Hystrix事件源与事件流：HystrixEvent和HystrixEventStream

public abstract class HystrixCommandEvent implements HystrixEvent {

    private final HystrixCommandKey commandKey;
    private final HystrixThreadPoolKey threadPoolKey;
    ... // 省略构造器个属性的get方法

	// command是否开始
    public abstract boolean isExecutionStart();
    // 是否是在隔离线程里执行的（先线程池模式隔离）
    public abstract boolean isExecutedInThread();
    // 响应被线程池拒绝
    public abstract boolean isResponseThreadPoolRejected();
    // command是否执行完成（包括成功、错误等）
    public abstract boolean isCommandCompletion();
    // command命令是否执行过
    public abstract boolean didCommandExecute();

	// 提供两个public的功能函数，过滤
	// 过滤只需要已经完成了的HystrixCommandEvent数据
	// 过滤只需要执行过的数据	
	public static final Func1<HystrixCommandEvent, Boolean> filterCompletionsOnly = (commandEvent) -> commandEvent.isCommandCompletion();
	public static final Func1<HystrixCommandEvent, Boolean> filterActualExecutions = (commandEvent) -> commandEvent.didCommandExecute();
}

public class HystrixCommandExecutionStarted extends HystrixCommandEvent {
	// 开始执行的时候必须告诉我隔离策略，以及当前并发数
	private final HystrixCommandProperties.ExecutionIsolationStrategy isolationStrategy;
	private final int currentConcurrency;
	... // 省略构造器

	//=====实现抽象类方法======
    @Override
    public boolean isExecutionStart() {
        return true;
    }
    @Override
    public boolean isExecutedInThread() {
        return isolationStrategy == HystrixCommandProperties.ExecutionIsolationStrategy.THREAD;
    }
    ... // 其它几个方法很明显，都是返回固定值：false
}

public class HystrixCommandCompletion extends HystrixCommandEvent {
	
	// 执行结果封装
    protected final ExecutionResult executionResult;
    // Hystrix的请求上下文。上上篇文章详细描述过
    // 默认会取值：HystrixRequestContext.getContextForCurrentThread()
    protected final HystrixRequestContext requestContext;
    // 所有的源生事件类型
	private final static HystrixEventType[] ALL_EVENT_TYPES = HystrixEventType.values();
	... // 省略构造方法
	
    @Override
    public boolean isExecutionStart() {
        return false;
    }
    ...
    @Override
    public boolean didCommandExecute() {
        return executionResult.executionOccurred();
    }
    ... // 生路其它get方法
    public HystrixRequestContext getRequestContext() {
        return this.requestContext;
    }
}

public class ExecutionResult {

    private final EventCounts eventCounts;
    private final Exception failedExecutionException;
    private final Exception executionException;
    private final long startTimestamp;
    private final int executionLatency;
    private final int userThreadLatency;
    private final boolean executionOccurred;
    private final boolean isExecutedInThread;
    private final HystrixCollapserKey collapserKey;
    ... 
}

// Observable表示数据发射器
public interface HystrixEventStream<E extends HystrixEvent> {
    Observable<E> observe();
}

public class HystrixCommandStartStream implements HystrixEventStream<HystrixCommandExecutionStarted> {

	private final HystrixCommandKey commandKey;
	// Subject它既能发送，又能监听
	// 发送和接受的数据类型是一样，均是HystrixCommandExecutionStarted类型
	private final Subject<HystrixCommandExecutionStarted, HystrixCommandExecutionStarted> writeOnlySubject;
	private final Observable<HystrixCommandExecutionStarted> readOnlyStream;

	// 缓存：每个commandKey对应同一个HystrixCommandStartStream实例
	// 这样发送数据流也方便统计
	private static final ConcurrentMap<String, HystrixCommandStartStream> streams = new ConcurrentHashMap<>();

	// 获取commandKey对应的Stream发射器实例
	public static HystrixCommandStartStream getInstance(HystrixCommandKey commandKey) {
		... // 先查缓存，没有就构造一个放进缓存
	}
    HystrixCommandStartStream(final HystrixCommandKey commandKey) {
        this.commandKey = commandKey;

        this.writeOnlySubject = new SerializedSubject(PublishSubject.create());
        // 这个只读的（共享的）流非常有意思，下面有介绍
        this.readOnlyStream = writeOnlySubject.share();
    }
    public static void reset() {
        streams.clear();
    }


	// 重要。提供写方法：把该event写到发射器里面去，这样订阅者就能读啦
	// 该方法的唯一调用处是：HystrixThreadEventStream
    public void write(HystrixCommandExecutionStarted event) {
        writeOnlySubject.onNext(event);
    }

	// 获取Observable对象.它是只读的，并不能发射数据哦
	// 但是你可以对它做流式处理，如.window.flatMap.share()...
    @Override
    public Observable<HystrixCommandExecutionStarted> observe() {
        return readOnlyStream;
    }
    ...
}

@Test
public void fun1() {
    HystrixCommandKey commandKey = HystrixCommandKey.Factory.asKey("demo");
    HystrixThreadPoolKey threadPoolKey = HystrixThreadPoolKey.Factory.asKey("demoThreadPool");
    HystrixCommandProperties.ExecutionIsolationStrategy isolationStrategy = HystrixCommandProperties.ExecutionIsolationStrategy.THREAD;

    HystrixCommandStartStream startStream = HystrixCommandStartStream.getInstance(commandKey);

    // 注册监听者
    startStream.observe()
            // .subscribeOn(Schedulers.io())
            // .observeOn(Schedulers.io())
            .observeOn(Schedulers.immediate())
            .subscribe(new Subscriber<HystrixCommandExecutionStarted>() {

                @Override
                public void onCompleted() {
                    System.out.println("数据发射完成啦");
                }

                @Override
                public void onError(Throwable e) {
                    System.out.println("数据发射出错啦：" + e.getMessage());
                }

                @Override
                public void onNext(HystrixCommandExecutionStarted hystrixCommand) {
                    System.out.printf("线程[%s] 数据发射start：%s %s %s %s",
                            Thread.currentThread().getName(),
                            hystrixCommand.getCommandKey(),
                            hystrixCommand.getThreadPoolKey(),
                            hystrixCommand.isExecutedInThread(),
                            hystrixCommand.getCurrentConcurrency());
                }
            });


    // 写数据：会马上发射出去
    startStream.write(new HystrixCommandExecutionStarted(commandKey, threadPoolKey, isolationStrategy, 6));
}

线程[main] 数据发射start：demo demoThreadPool true 6