Turbine原理

Turbine架构

Concept

clusters:a group of servers that need to be monitored together. Turbine aggregates all metrics from a cluster of servers together, hence when you request the aggregate feed from Turbine, you need to specify the cluster query param.
name+type:Note that Turbine needs an aggregation dimension when combining data from multiple {@link DataFromSingleInstance} objects into a single {@link AggDataFromCluster} object.The name and type are used for this purpose. name + type is the unique aggregation key. name identifies a single data point / event / metric etc whereas type is used to denote the sub stream type which then allows us to multiplex multiple sub streams over the same connection from an instance.

name and type can also be specified as filtering criteria when fetching the output stream from the turbine aggregator.

重要组件

PluginsFactory只包含两个对象的引用InstanceDiscovery和ClusterMonitorFactory；
1. TurbineData：有DataFromSingleInstance和AggDataFromCluster两个实现，一个DataFromSingleInstance对应一个InstanceMonitor对象，一个AggDataFromCluster对应一个ClusterMonitor对象；
2. InstanceUrlClosure：根据Instance中对象生成hystrix.stream的完整URL地址；
3. InstanceObservable：一个Instance一个InstanceObservable对象，实时监控Instance的状态；hostUp时创建一个InstanceMonitor
4. TurbineDataHandler：真正处理数据的地方；TurbineDataDispatcher中，dispach的意义就是将数据发送给系统中所有的TurbineDataHandler对象，让TurbineDataHandler对象真正的处理数据；
3. TurbineDataDispatcher：收到的数据分发给对应的TurbineDataHandler；
4. TurbineDataMonitor：InstanceMonitor实时监控instance的状态，其实就是一致调用hystrix.stream接口，请求回来的数据调用TurbineDataDispatcher的pushData方法，进行分发，最终是调用系统中所有的TurbineDataHandler对象实例来处理数据；
5. MonitorConsole： 多个TurbineDataMonitor的集合，典型的用法是，ClusterMonitor通过此对象持有此集群下的所有的InstanceMonitor对象；

Thread

InstanceObservable

InstanceObservable是个单例对象，在Turbine.init()方法调用中调用了InstanceObservable的start方法启动，启动之后，会定时调度一个TimerTask对象，这个TimerTask对象的操作是，
1. 实时获取集群中所有的Instance的当前状态，注意是Instance，不是Service；
2. 根据此次获取到的所有的instance的状态，更新系统中相关的数据，比如当前当机了多少
3. 循环回调所有的InstanceObserver对象的hostUps和hostDowns方法
4. updateHostsCountsPerCluster(HashSet hostsUp)

InstanceMonitor

InstanceMonitor#startMonitor()方法会启动启动一个线程，此线程的功能是请求此实例的hystrix.stream接口，生成最新的DataFromSingleInstance对象；调用TurbineDataDispatcher#pushData方法真正的处理数据；

ClusterMonitor

ClusterMonitor的实现是AggregateClusterMonitor，AggregateClusterMonitor#startMonitor()中会调用StaleConnectionMonitorReaper的start()方法，StaleConnectionMonitorReaper#start()中会启动一个线程，此线程的主要工作是checkStaleMonitors，对当前AggregateClusterMonitor中管理的所有的InstanceMonitor的健康检测，如果这个InstanceMonitor所链接的远程服务超过一定的时间间隔没有响应hystrix.stream请求，就将instanceMonitor #stopMonitor()；

Spring Cloud Turbine

Configuration

1. turbine.appConfig是服务的名称列表，服务是Eureka中ServiceId的概念，一个服务会有多个实例；
2. clusterNameExpression，获取ClusterName的表达式；

Implemention

1. 重写了Spring版本的ClusterMonitorFactory和Eureka版本的InstanceDiscovery接口实现；
2. InstanceUrlClosure提供了对HTTPS的支持；
3. 提供Spring Boot风格的配置时用方式；

参考

Hystrix-Turbine 配置及git源码路径
Turbine源码分析