02、Hystrix实战：Hystrix之metrics

**　　**上一节讲到了hystrix提供的五个功能，这一节我们首先来讲hystrix中提供实时执行metrics信息的实现。为什么先讲metrics，因为很多功能都是基于metrics的数据来实现的，它是很多功能实现的基础。

　　首先来看一下通过hystrix调用服务的过程中会产生那些类型的metrics信息：

　　1.某一事件的持续指标。

　　2.某一事件窗口时间内持续指标。

　　3.某一事件窗口时间内最大指标。

　　4.某一事件窗口时间内指标分布。

　　在来看一下这些数据在hystrix中是如何产生、计算和流转的。

　　hystrix在执行服务调用的过程中会产生各类事件，执行模块首先将这些事件发送的metrics接受流中，而metrics统计流会监听metrics接受流，计算出各类统计数据。

metrics接收流

　　hystrix有以下接收流和对应接收的消息

　　metrics接收流使用单例模式，HystrixCommandKey，HystrixThreadPoolKey，HystrixCollapserKey分别对应同一个（HystrixCommandStartStream、HystrixCommandCompletionStream），（HystrixThreadPoolStartStream，HystrixThreadPoolCompletionStream），（HystrixCollapserEventStream）。

　　内部使用rxjava来实现消息机制

    HystrixCommandStartStream(final HystrixCommandKey commandKey) {
        this.commandKey = commandKey;
        this.writeOnlySubject = new SerializedSubject<HystrixCommandExecutionStarted, HystrixCommandExecutionStarted>(PublishSubject.<HystrixCommandExecutionStarted>create());
        this.readOnlyStream = writeOnlySubject.share();
    }

**　　**此外还提供了HystrixThreadEventStream统一执行接收消息然后发送到各个消息接收流类。

metrics接受流消息体

　　上面讲了hystrix metrics的接收流，接下来我们看看接收流具体接收的内容。

```java

    private final EventCounts eventCounts;／／事件数量
private final Exception failedExecutionException;//失败异常
private final Exception executionException; //执行异常
private final long startTimestamp;//命令开始执行时间
private final int executionLatency; //执行run的时间
private final int userThreadLatency; //请求提交到执行结束的时间
private final boolean executionOccurred;//ture 执行过命令 false 未执行过命令
private final boolean isExecutedInThread;//ture 使用线程池执行 false 不是使用线程池执行
```

```java

   private final BitSet events;事件类型
   private final int numEmissions//emission次数
   private final int numFallbackEmissions;//fallback次数
```

```java

private final HystrixCollapserKey collapserKey;//合并命令key
```

事件类型：

　　HystrixCommand只返回一个数据，当返回值时发生SUCCESS事件，执行失败时，发生FAILURE事件，HystrixObservableCommand可以返回多个值，当返回值时发生EMIT事件，当命令完成时，发生SUCCESS事件，执行失败时，发生FAILURE事件。

Fallback事件类型

其他命令类型

线程池类型

聚合事件类型

metrics统计流

　　hystrix有以下统计流

　　窗口时间内持续统计流首先监听一个消息接受流，统计一段时间内各个类型消息的累计数据（时间为：metrics.rollingStats.timeInMilliseconds／metrics.rollingStats.numBuckets）。然后再对累计的数据进行累加（个数为：metrics.rollingStats.numBuckets），即为最终累计数据。

　　持续统计流首先监听一个消息流（开始消息流或者完成消息流），统计一段时间内各个类型消息的累计数据（时间为：metrics.rollingStats.timeInMilliseconds／metrics.rollingStats.numBuckets）。然后不断的累加累计数据。

　　窗口时间内分布统计流首先监听一个消息流，统计一段时间内各个类型消息存放在Histogram对象中（时间为：metrics.rollingStats.timeInMilliseconds／metrics.rollingStats.numBuckets），然后对（个数为：metrics.rollingStats.numBuckets）内的Histogram对象进行运算操作，即为窗口期内某一时间的分布。

　　RollingConcurrencyStream监听一个消息流，例如HystrixCommandStartStream，然后通过RX java对一段时间内的执行并发量取最大值，重新发射，对窗口期内的执行并发量取最大值，重新发射。

　　metrics统计流使用单例模式，每个统计流分别对应一个HystrixCommandKey，HystrixThreadPoolKey，HystrixCollapserKey。

metrics模块

　　hystrix中可以通过metrics模块来获取执行过程中的数据，主要有三部分数据：命令执行metrics，线程池metrics，合并命令执行metrics，每个HystrixCommandKey、HystrixThreadPoolKey、HystrixCollapserKey对应一个相应的metrics（HystrixCommandMetrics，HystrixThreadPoolMetrics,HystrixCollapserMetrics）。metrics模块内部是通过监听消息流来获取各个指标的统计数据。

命令执行metrics

　　HystrixCommandMetrics为命令执行模块的metrics，在其初始化时会创建6个统计数据流：HealthCountsStream、RollingCommandEventCounterStream、CumulativeCommandEventCounterStream、RollingCommandLatencyDistributionStream、RollingCommandUserLatencyDistributionStream、RollingCommandMaxConcurrencyStream，通过这些统计数据流来获取相应metrics信息。

private HealthCountsStream healthCountsStream;
    private final RollingCommandEventCounterStream rollingCommandEventCounterStream;
    private final CumulativeCommandEventCounterStream cumulativeCommandEventCounterStream;
    private final RollingCommandLatencyDistributionStream rollingCommandLatencyDistributionStream;
    private final RollingCommandUserLatencyDistributionStream rollingCommandUserLatencyDistributionStream;
    private final RollingCommandMaxConcurrencyStream rollingCommandMaxConcurrencyStream;
    /* package */HystrixCommandMetrics(final HystrixCommandKey key, HystrixCommandGroupKey commandGroup, HystrixThreadPoolKey threadPoolKey, HystrixCommandProperties properties, HystrixEventNotifier eventNotifier) {
        super(null);
        healthCountsStream = HealthCountsStream.getInstance(key, properties);
        rollingCommandEventCounterStream = RollingCommandEventCounterStream.getInstance(key, properties);
        cumulativeCommandEventCounterStream = CumulativeCommandEventCounterStream.getInstance(key, properties);
        rollingCommandLatencyDistributionStream = RollingCommandLatencyDistributionStream.getInstance(key, properties);
        rollingCommandUserLatencyDistributionStream = RollingCommandUserLatencyDistributionStream.getInstance(key, properties);
        rollingCommandMaxConcurrencyStream = RollingCommandMaxConcurrencyStream.getInstance(key, properties);
    }
　　//获取指定事件窗口期内数据指标
    public long getRollingCount(HystrixEventType eventType) {
        return rollingCommandEventCounterStream.getLatest(eventType);
    }
　　//获取指定事件持续的数据指标
    public long getCumulativeCount(HystrixEventType eventType) {
        return cumulativeCommandEventCounterStream.getLatest(eventType);
    }//获取某一百分比的执行时间public int getExecutionTimePercentile(double percentile) {
        return rollingCommandLatencyDistributionStream.getLatestPercentile(percentile);
    }//获取平均的执行时间
    public int getExecutionTimeMean() {
        return rollingCommandLatencyDistributionStream.getLatestMean();
    }
    //获取某一百分比的总时间
    public int getTotalTimePercentile(double percentile) {
        return rollingCommandUserLatencyDistributionStream.getLatestPercentile(percentile);
    }//获取平均的总时间
    public int getTotalTimeMean() {
        return rollingCommandUserLatencyDistributionStream.getLatestMean();
    }
    //获取窗口期内最大并发量
    public long getRollingMaxConcurrentExecutions() {
        return rollingCommandMaxConcurrencyStream.getLatestRollingMax();
    }//获取当前并发量
    public int getCurrentConcurrentExecutionCount() {
        return concurrentExecutionCount.get();
    }  //获取命令执行健康情况
    public HealthCounts getHealthCounts() {
        return healthCountsStream.getLatest();
    }

线程池metrics

　　HystrixThreadPoolMetrics为线程池执行模块的metrics，在其初始化时会获取3个数据流：RollingThreadPoolEventCounterStream、CumulativeThreadPoolEventCounterStream、RollingThreadPoolMaxConcurrencyStream通过这些统计流获得相应的统计数据。

    private final RollingThreadPoolEventCounterStream rollingCounterStream;
    private final CumulativeThreadPoolEventCounterStream cumulativeCounterStream;
    private final RollingThreadPoolMaxConcurrencyStream rollingThreadPoolMaxConcurrencyStream;
    private HystrixThreadPoolMetrics(HystrixThreadPoolKey threadPoolKey, ThreadPoolExecutor threadPool, HystrixThreadPoolProperties properties) {
        super(null);
        this.threadPoolKey = threadPoolKey;
        this.threadPool = threadPool;
        this.properties = properties;
        rollingCounterStream = RollingThreadPoolEventCounterStream.getInstance(threadPoolKey, properties);
        cumulativeCounterStream = CumulativeThreadPoolEventCounterStream.getInstance(threadPoolKey, properties);
        rollingThreadPoolMaxConcurrencyStream = RollingThreadPoolMaxConcurrencyStream.getInstance(threadPoolKey, properties);
    }
/**
     获取窗口期内线程池执行的个数*/
    public long getRollingCountThreadsExecuted() {
        return rollingCounterStream.getLatestCount(HystrixEventType.ThreadPool.EXECUTED);
    }

    /**
     获取持续的线程池执行个数*/
    public long getCumulativeCountThreadsExecuted() {
        return cumulativeCounterStream.getLatestCount(HystrixEventType.ThreadPool.EXECUTED);
    }

    /**
    获取窗口期内线程池拒绝的个数*/
    public long getRollingCountThreadsRejected() {
        return rollingCounterStream.getLatestCount(HystrixEventType.ThreadPool.REJECTED);
    }

    /**
    获取持续内线程池拒绝的个数*/
    public long getCumulativeCountThreadsRejected() {
        return cumulativeCounterStream.getLatestCount(HystrixEventType.ThreadPool.REJECTED);
    }
    //获取指定事件窗口期内数据指标
    public long getRollingCount(HystrixEventType.ThreadPool event) {
        return rollingCounterStream.getLatestCount(event);
    }
   //获取指定事件持续的数据指标
    public long getCumulativeCount(HystrixEventType.ThreadPool event) {
        return cumulativeCounterStream.getLatestCount(event);
    }/**
    获取窗口期内最大并发量*/
    public long getRollingMaxActiveThreads() {
        return rollingThreadPoolMaxConcurrencyStream.getLatestRollingMax();
    }

还有一些根据线程池获取线程池当前指标

    public Number getCurrentActiveCount() {
        return threadPool.getActiveCount();
    }
    public Number getCurrentCompletedTaskCount() {
        return threadPool.getCompletedTaskCount();
    }
    public Number getCurrentCorePoolSize() {
        return threadPool.getCorePoolSize();
    }
    public Number getCurrentLargestPoolSize() {
        return threadPool.getLargestPoolSize();
    }
    public Number getCurrentMaximumPoolSize() {
        return threadPool.getMaximumPoolSize();
    }
    public Number getCurrentPoolSize() {
        return threadPool.getPoolSize();
    }
    public Number getCurrentTaskCount() {
        return threadPool.getTaskCount();
    }
    public Number getCurrentQueueSize() {
        return threadPool.getQueue().size();
    }

合并命令执行metrics

**　　**HystrixCollapserMetrics为合并命令执行模块的metrics，在其初始化时会创建3个数据流：RollingCollapserEventCounterStream、CumulativeCollapserEventCounterStream、RollingCollapserBatchSizeDistributionStream，通过这些统计流获得相应的统计数据。

    private final RollingCollapserEventCounterStream rollingCollapserEventCounterStream;
    private final CumulativeCollapserEventCounterStream cumulativeCollapserEventCounterStream;
    private final RollingCollapserBatchSizeDistributionStream rollingCollapserBatchSizeDistributionStream;

    /* package */HystrixCollapserMetrics(HystrixCollapserKey key, HystrixCollapserProperties properties) {
        super(null);
        rollingCollapserEventCounterStream = RollingCollapserEventCounterStream.getInstance(key, properties);
        cumulativeCollapserEventCounterStream = CumulativeCollapserEventCounterStream.getInstance(key, properties);
        rollingCollapserBatchSizeDistributionStream = RollingCollapserBatchSizeDistributionStream.getInstance(key, properties);
    }
    //获取指定事件窗口期内数据指标
    public long getRollingCount(HystrixEventType.Collapser collapserEventType) {
        return rollingCollapserEventCounterStream.getLatest(collapserEventType);
    }
    //获取指定事件持续的数据指标
    public long getCumulativeCount(HystrixEventType.Collapser collapserEventType) {
        return cumulativeCollapserEventCounterStream.getLatest(collapserEventType);
    }
    //获取指定百分比的batchsize
    public int getBatchSizePercentile(double percentile) {
        return rollingCollapserBatchSizeDistributionStream.getLatestPercentile(percentile);
    }
    //获取平均的batchsize
    public int getBatchSizeMean() {
        return rollingCollapserBatchSizeDistributionStream.getLatestMean();
    }

其他流

HystrixConfigurationStream

　　该数据流定时将hystrix的最新properties配置，发送到该消息流中。com.netflix.hystrix.contrib.sample.stream.HystrixConfigSseServlet就是用该流来获取配置信息。

public HystrixConfigurationStream(final int intervalInMilliseconds) {
        this.intervalInMilliseconds = intervalInMilliseconds;
        this.allConfigurationStream = Observable.interval(intervalInMilliseconds, TimeUnit.MILLISECONDS)
                .map(getAllConfig)
                .doOnSubscribe(new Action0() {
                    @Override
                    public void call() {
                        isSourceCurrentlySubscribed.set(true); } }) .doOnUnsubscribe(new Action0() { @Override public void call() { isSourceCurrentlySubscribed.set(false); } }) .share() .onBackpressureDrop(); } private static final Func1<Long, HystrixConfiguration> getAllConfig = new Func1<Long, HystrixConfiguration>() { @Override public HystrixConfiguration call(Long timestamp) { return HystrixConfiguration.from( getAllCommandConfig.call(timestamp), getAllThreadPoolConfig.call(timestamp), getAllCollapserConfig.call(timestamp) ); } };

private static final Func1<Long, Map<HystrixCommandKey, HystrixCommandConfiguration>> getAllCommandConfig =
            new Func1<Long, Map<HystrixCommandKey, HystrixCommandConfiguration>>() {
                @Override
                public Map<HystrixCommandKey, HystrixCommandConfiguration> call(Long timestamp) {
                    Map<HystrixCommandKey, HystrixCommandConfiguration> commandConfigPerKey = new HashMap<HystrixCommandKey, HystrixCommandConfiguration>();
                    for (HystrixCommandMetrics commandMetrics: HystrixCommandMetrics.getInstances()) {
                        HystrixCommandKey commandKey = commandMetrics.getCommandKey();
                        HystrixThreadPoolKey threadPoolKey = commandMetrics.getThreadPoolKey(); HystrixCommandGroupKey groupKey = commandMetrics.getCommandGroup(); commandConfigPerKey.put(commandKey, sampleCommandConfiguration(commandKey, threadPoolKey, groupKey, commandMetrics.getProperties())); } return commandConfigPerKey; } }; private static final Func1<Long, Map<HystrixThreadPoolKey, HystrixThreadPoolConfiguration>> getAllThreadPoolConfig = new Func1<Long, Map<HystrixThreadPoolKey, HystrixThreadPoolConfiguration>>() { @Override public Map<HystrixThreadPoolKey, HystrixThreadPoolConfiguration> call(Long timestamp) { Map<HystrixThreadPoolKey, HystrixThreadPoolConfiguration> threadPoolConfigPerKey = new HashMap<HystrixThreadPoolKey, HystrixThreadPoolConfiguration>(); for (HystrixThreadPoolMetrics threadPoolMetrics: HystrixThreadPoolMetrics.getInstances()) { HystrixThreadPoolKey threadPoolKey = threadPoolMetrics.getThreadPoolKey(); threadPoolConfigPerKey.put(threadPoolKey, sampleThreadPoolConfiguration(threadPoolKey, threadPoolMetrics.getProperties())); } return threadPoolConfigPerKey; } }; private static final Func1<Long, Map<HystrixCollapserKey, HystrixCollapserConfiguration>> getAllCollapserConfig = new Func1<Long, Map<HystrixCollapserKey, HystrixCollapserConfiguration>>() { @Override public Map<HystrixCollapserKey, HystrixCollapserConfiguration> call(Long timestamp) { Map<HystrixCollapserKey, HystrixCollapserConfiguration> collapserConfigPerKey = new HashMap<HystrixCollapserKey, HystrixCollapserConfiguration>(); for (HystrixCollapserMetrics collapserMetrics: HystrixCollapserMetrics.getInstances()) { HystrixCollapserKey collapserKey = collapserMetrics.getCollapserKey(); collapserConfigPerKey.put(collapserKey, sampleCollapserConfiguration(collapserKey, collapserMetrics.getProperties())); } return collapserConfigPerKey; } };

metrics发布

　　有时，我们需要发布Hystrix中的metrics到其他地方，Hystrix提供了相应的接口（HystrixMetricsPublisherCollapser，HystrixMetricsPublisherCommand，HystrixMetricsPublisherThreadPool），实现这些接口，并在initial方法中实现发送hystrix的metrics的逻辑。实现HystrixMetricsPublisher，来创建这些实现类。其实hystrix对metrics的发布只是定义了接口和initial方法。Hystrix运行时，HystrixMetricsPublisherFactory通过HystrixPlugins获取HystrixMetricsPublisher的实现类。并且通过该实现类来创建（HystrixMetricsPublisherCollapser，HystrixMetricsPublisherCommand，HystrixMetricsPublisherThreadPool）的实现类，并在初次创建时调用其initial方法。

例如：

　　通过coda hale 实现了将hystrix 的metrics信息输出到指定metrics监控系统中。

　　引入jar包：

　　　　　<dependency>
            <groupId>com.netflix.hystrix</groupId>
            <artifactId>hystrix-codahale-metrics-publisher</artifactId>
            <version>1.5.9</version>
        </dependency>

　　创建HystrixMetricsPublisher对象并注册到HystrixPlugins：

　　@Bean

HystrixMetricsPublisher hystrixMetricsPublisher() {

HystrixCodaHaleMetricsPublisher publisher = new HystrixCodaHaleMetricsPublisher(metricRegistry);

HystrixPlugins.getInstance().registerMetricsPublisher(publisher);

return publisher;

}

　　coda hale实现源码如下：

public class HystrixCodaHaleMetricsPublisher extends HystrixMetricsPublisher {

　　    private final String metricsRootNode;

    　　private final MetricRegistry metricRegistry;

    　　public HystrixCodaHaleMetricsPublisher(MetricRegistry metricRegistry) {

        　　this(null, metricRegistry);

    　　}

    　　public HystrixCodaHaleMetricsPublisher(String metricsRootNode, MetricRegistry metricRegistry) {

       　　 this.metricsRootNode = metricsRootNode;

        　　this.metricRegistry = metricRegistry;

    　　}

    　　@Override

    　　public HystrixMetricsPublisherCommand getMetricsPublisherForCommand(HystrixCommandKey commandKey, HystrixCommandGroupKey commandGroupKey, HystrixCommandMetrics metrics, HystrixCircuitBreaker circuitBreaker, HystrixCommandProperties properties) {

        　　return new HystrixCodaHaleMetricsPublisherCommand(metricsRootNode, commandKey, commandGroupKey, metrics, circuitBreaker, properties, metricRegistry);

    　　}

    　　@Override

    　　public HystrixMetricsPublisherThreadPool getMetricsPublisherForThreadPool(HystrixThreadPoolKey threadPoolKey, HystrixThreadPoolMetrics metrics, HystrixThreadPoolProperties properties) {

        　　return new HystrixCodaHaleMetricsPublisherThreadPool(metricsRootNode, threadPoolKey, metrics, properties, metricRegistry);

    　　}

    　　@Override

    　　public HystrixMetricsPublisherCollapser getMetricsPublisherForCollapser(HystrixCollapserKey collapserKey, HystrixCollapserMetrics metrics, HystrixCollapserProperties properties) {

        　　return new HystrixCodaHaleMetricsPublisherCollapser(collapserKey, metrics, properties, metricRegistry);

    　　}

　　}

　　HystrixCodaHaleMetricsPublisher负责创建HystrixCodaHaleMetricsPublisherCommand，HystrixCodaHaleMetricsPublisherThreadPool，HystrixCodaHaleMetricsPublisherCollapser。这三个对象实现基本逻辑是在initialize方法中向metricRegistry中设置相应信息。

public void initialize() {    　　metricRegistry.register(createMetricName("isCircuitBreakerOpen"), new Gauge<Boolean>() {

    　　@Override

    　　public Boolean getValue() {

         　　return circuitBreaker.isOpen();

    　　}

　   .....

 }