Hystrix使用詳解

原文參考：http://hot66hot.iteye.com/blog/2155036

 

一:為什么需要Hystrix?

在大中型分布式系統中，通常系統很多依賴(HTTP,hession,Netty,Dubbo等)，如下圖:

 

在高並發訪問下,這些依賴的穩定性與否對系統的影響非常大,但是依賴有很多不可控問題:如網絡連接緩慢，資源繁忙，暫時不可用，服務脫機等.

如下圖：QPS為50的依賴 I 出現不可用，但是其他依賴仍然可用.

 

當依賴I 阻塞時,大多數服務器的線程池就出現阻塞(BLOCK),影響整個線上服務的穩定性.如下圖:

 

在復雜的分布式架構的應用程序有很多的依賴，都會不可避免地在某些時候失敗。高並發的依賴失敗時如果沒有隔離措施，當前應用服務就有被拖垮的風險。

 

例如:一個依賴30個SOA服務的系統,每個服務99.99%可用。
99.99%的30次方 ≈ 99.7%
0.3% 意味着一億次請求 會有 3,000,00次失敗
換算成時間大約每月有2個小時服務不穩定.
隨着服務依賴數量的變多，服務不穩定的概率會成指數性提高.

 解決問題方案:對依賴做隔離,Hystrix就是處理依賴隔離的框架,同時也是可以幫我們做依賴服務的治理和監控.

 

Netflix 公司開發並成功使用Hystrix,使用規模如下:

 

The Netflix API processes 10+ billion HystrixCommand executions per day using thread isolation.
Each API instance has 40+ thread-pools with 5-20 threads in each (most are set to 10).

二:Hystrix如何解決依賴隔離

1:Hystrix使用命令模式HystrixCommand(Command)包裝依賴調用邏輯，每個命令在單獨線程中/信號授權下執行。

2:可配置依賴調用超時時間,超時時間一般設為比99.5%平均時間略高即可.當調用超時時，直接返回或執行fallback邏輯。

3:為每個依賴提供一個小的線程池（或信號），如果線程池已滿調用將被立即拒絕，默認不采用排隊.加速失敗判定時間。

4:依賴調用結果分:成功，失敗（拋出異常），超時，線程拒絕，短路。 請求失敗(異常，拒絕，超時，短路)時執行fallback(降級)邏輯。

5:提供熔斷器組件,可以自動運行或手動調用,停止當前依賴一段時間(10秒)，熔斷器默認錯誤率閾值為50%,超過將自動運行。

6:提供近實時依賴的統計和監控

Hystrix依賴的隔離架構,如下圖:

三:如何使用Hystrix

1:使用maven引入Hystrix依賴

 

<!-- 依賴版本 -->
<hystrix.version>1.3.16</hystrix.version>
<hystrix-metrics-event-stream.version>1.1.2</hystrix-metrics-event-stream.version> 
 
<dependency>
     <groupId>com.netflix.hystrix</groupId>
     <artifactId>hystrix-core</artifactId>
     <version>${hystrix.version}</version>
 </dependency>
     <dependency>
     <groupId>com.netflix.hystrix</groupId>
     <artifactId>hystrix-metrics-event-stream</artifactId>
     <version>${hystrix-metrics-event-stream.version}</version>
 </dependency>
<!-- 倉庫地址 -->
<repository>
     <id>nexus</id>
     <name>local private nexus</name>
     <url>http://maven.oschina.net/content/groups/public/</url>
     <releases>
          <enabled>true</enabled>
     </releases>
     <snapshots>
          <enabled>false</enabled>
     </snapshots>
</repository>

 

2:使用命令模式封裝依賴邏輯

 

public class HelloWorldCommand extends HystrixCommand<String> {
    private final String name;
    public HelloWorldCommand(String name) {
        //最少配置:指定命令組名(CommandGroup)
        super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"));
        this.name = name;
    }
    @Override
    protected String run() {
        // 依賴邏輯封裝在run()方法中
        return "Hello " + name +" thread:" + Thread.currentThread().getName();
    }
    //調用實例
    public static void main(String[] args) throws Exception{
        //每個Command對象只能調用一次,不可以重復調用,
        //重復調用對應異常信息:This instance can only be executed once. Please instantiate a new instance.
        HelloWorldCommand helloWorldCommand = new HelloWorldCommand("Synchronous-hystrix");
        //使用execute()同步調用代碼,效果等同於:helloWorldCommand.queue().get(); 
        String result = helloWorldCommand.execute();
        System.out.println("result=" + result);
 
        helloWorldCommand = new HelloWorldCommand("Asynchronous-hystrix");
        //異步調用,可自由控制獲取結果時機,
        Future<String> future = helloWorldCommand.queue();
        //get操作不能超過command定義的超時時間,默認:1秒
        result = future.get(100, TimeUnit.MILLISECONDS);
        System.out.println("result=" + result);
        System.out.println("mainThread=" + Thread.currentThread().getName());
    }
     
}
    //運行結果: run()方法在不同的線程下執行
    // result=Hello Synchronous-hystrix thread:hystrix-HelloWorldGroup-1
    // result=Hello Asynchronous-hystrix thread:hystrix-HelloWorldGroup-2
    // mainThread=main

 

 note:異步調用使用 command.queue()get(timeout, TimeUnit.MILLISECONDS);同步調用使用command.execute() 等同於 command.queue().get();

3:注冊異步事件回調執行

//注冊觀察者事件攔截
Observable<String> fs = new HelloWorldCommand("World").observe();
//注冊結果回調事件
fs.subscribe(new Action1<String>() {
    @Override
    public void call(String result) {
         //執行結果處理,result 為HelloWorldCommand返回的結果
        //用戶對結果做二次處理.
    }
});
//注冊完整執行生命周期事件
fs.subscribe(new Observer<String>() {
            @Override
            public void onCompleted() {
                // onNext/onError完成之后最后回調
                System.out.println("execute onCompleted");
            }
            @Override
            public void onError(Throwable e) {
                // 當產生異常時回調
                System.out.println("onError " + e.getMessage());
                e.printStackTrace();
            }
            @Override
            public void onNext(String v) {
                // 獲取結果后回調
                System.out.println("onNext: " + v);
            }
        });
/* 運行結果
call execute result=Hello observe-hystrix thread:hystrix-HelloWorldGroup-3
onNext: Hello observe-hystrix thread:hystrix-HelloWorldGroup-3
execute onCompleted
*/

 

4:使用Fallback() 提供降級策略

 

//重載HystrixCommand 的getFallback方法實現邏輯
public class HelloWorldCommand extends HystrixCommand<String> {
    private final String name;
    public HelloWorldCommand(String name) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("HelloWorldGroup"))
                /* 配置依賴超時時間,500毫秒*/
                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter().withExecutionIsolationThreadTimeoutInMilliseconds(500)));
        this.name = name;
    }
    @Override
    protected String getFallback() {
        return "exeucute Falled";
    }
    @Override
    protected String run() throws Exception {
        //sleep 1 秒,調用會超時
        TimeUnit.MILLISECONDS.sleep(1000);
        return "Hello " + name +" thread:" + Thread.currentThread().getName();
    }
    public static void main(String[] args) throws Exception{
        HelloWorldCommand command = new HelloWorldCommand("test-Fallback");
        String result = command.execute();
    }
}
/* 運行結果:getFallback() 調用運行
getFallback executed
*/

 

NOTE: 除了HystrixBadRequestException異常之外，所有從run()方法拋出的異常都算作失敗，並觸發降級getFallback()和斷路器邏輯。

          HystrixBadRequestException用在非法參數或非系統故障異常等不應觸發回退邏輯的場景。

5:依賴命名:CommandKey

public HelloWorldCommand(String name) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))
                /* HystrixCommandKey工廠定義依賴名稱 */
                .andCommandKey(HystrixCommandKey.Factory.asKey("HelloWorld")));
        this.name = name;
    }

 

 NOTE: 每個CommandKey代表一個依賴抽象,相同的依賴要使用相同的CommandKey名稱。依賴隔離的根本就是對相同CommandKey的依賴做隔離.

6:依賴分組:CommandGroup

命令分組用於對依賴操作分組,便於統計,匯總等.

//使用HystrixCommandGroupKey工廠定義
public HelloWorldCommand(String name) {
    Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("HelloWorldGroup"))
}

 

 NOTE: CommandGroup是每個命令最少配置的必選參數，在不指定ThreadPoolKey的情況下，字面值用於對不同依賴的線程池/信號區分.

7:線程池/信號:ThreadPoolKey

public HelloWorldCommand(String name) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))
                .andCommandKey(HystrixCommandKey.Factory.asKey("HelloWorld"))
                /* 使用HystrixThreadPoolKey工廠定義線程池名稱*/
                .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("HelloWorldPool")));
        this.name = name;
    }

 NOTE: 當對同一業務依賴做隔離時使用CommandGroup做區分,但是對同一依賴的不同遠程調用如(一個是redis 一個是http),可以使用HystrixThreadPoolKey做隔離區分.

           最然在業務上都是相同的組，但是需要在資源上做隔離時，可以使用HystrixThreadPoolKey區分.

8:請求緩存 Request-Cache

public class RequestCacheCommand extends HystrixCommand<String> {
    private final int id;
    public RequestCacheCommand( int id) {
        super(HystrixCommandGroupKey.Factory.asKey("RequestCacheCommand"));
        this.id = id;
    }
    @Override
    protected String run() throws Exception {
        System.out.println(Thread.currentThread().getName() + " execute id=" + id);
        return "executed=" + id;
    }
    //重寫getCacheKey方法,實現區分不同請求的邏輯
    @Override
    protected String getCacheKey() {
        return String.valueOf(id);
    }
 
    public static void main(String[] args){
        HystrixRequestContext context = HystrixRequestContext.initializeContext();
        try {
            RequestCacheCommand command2a = new RequestCacheCommand(2);
            RequestCacheCommand command2b = new RequestCacheCommand(2);
            Assert.assertTrue(command2a.execute());
            //isResponseFromCache判定是否是在緩存中獲取結果
            Assert.assertFalse(command2a.isResponseFromCache());
            Assert.assertTrue(command2b.execute());
            Assert.assertTrue(command2b.isResponseFromCache());
        } finally {
            context.shutdown();
        }
        context = HystrixRequestContext.initializeContext();
        try {
            RequestCacheCommand command3b = new RequestCacheCommand(2);
            Assert.assertTrue(command3b.execute());
            Assert.assertFalse(command3b.isResponseFromCache());
        } finally {
            context.shutdown();
        }
    }
}

 

 NOTE:請求緩存可以讓(CommandKey/CommandGroup)相同的情況下,直接共享結果，降低依賴調用次數，在高並發和CacheKey碰撞率高場景下可以提升性能.

Servlet容器中，可以直接實用Filter機制Hystrix請求上下文

public class HystrixRequestContextServletFilter implements Filter {
    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) 
     throws IOException, ServletException {
        HystrixRequestContext context = HystrixRequestContext.initializeContext();
        try {
            chain.doFilter(request, response);
        } finally {
            context.shutdown();
        }
    }
}
<filter>
      <display-name>HystrixRequestContextServletFilter</display-name>
      <filter-name>HystrixRequestContextServletFilter</filter-name>
      <filter-class>com.netflix.hystrix.contrib.requestservlet.HystrixRequestContextServletFilter</filter-class>
    </filter>
    <filter-mapping>
      <filter-name>HystrixRequestContextServletFilter</filter-name>
      <url-pattern>/*</url-pattern>
   </filter-mapping>

 

9:信號量隔離:SEMAPHORE

  隔離本地代碼或可快速返回遠程調用(如memcached,redis)可以直接使用信號量隔離,降低線程隔離開銷.

public class HelloWorldCommand extends HystrixCommand<String> {
    private final String name;
    public HelloWorldCommand(String name) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("HelloWorldGroup"))
                /* 配置信號量隔離方式,默認采用線程池隔離 */
                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter().withExecutionIsolationStrategy(HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE)));
        this.name = name;
    }
    @Override
    protected String run() throws Exception {
        return "HystrixThread:" + Thread.currentThread().getName();
    }
    public static void main(String[] args) throws Exception{
        HelloWorldCommand command = new HelloWorldCommand("semaphore");
        String result = command.execute();
        System.out.println(result);
        System.out.println("MainThread:" + Thread.currentThread().getName());
    }
}
/** 運行結果
 HystrixThread:main
 MainThread:main
*/

 

10:fallback降級邏輯命令嵌套

 

  適用場景:用於fallback邏輯涉及網絡訪問的情況,如緩存訪問。

 

public class CommandWithFallbackViaNetwork extends HystrixCommand<String> {
    private final int id;
 
    protected CommandWithFallbackViaNetwork(int id) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("RemoteServiceX"))
                .andCommandKey(HystrixCommandKey.Factory.asKey("GetValueCommand")));
        this.id = id;
    }
 
    @Override
    protected String run() {
        // RemoteService.getValue(id);
        throw new RuntimeException("force failure for example");
    }
 
    @Override
    protected String getFallback() {
        return new FallbackViaNetwork(id).execute();
    }
 
    private static class FallbackViaNetwork extends HystrixCommand<String> {
        private final int id;
        public FallbackViaNetwork(int id) {
            super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("RemoteServiceX"))
                    .andCommandKey(HystrixCommandKey.Factory.asKey("GetValueFallbackCommand"))
                    // 使用不同的線程池做隔離，防止上層線程池跑滿，影響降級邏輯.
                    .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("RemoteServiceXFallback")));
            this.id = id;
        }
        @Override
        protected String run() {
            MemCacheClient.getValue(id);
        }
 
        @Override
        protected String getFallback() {
            return null;
        }
    }
}

 

 NOTE:依賴調用和降級調用使用不同的線程池做隔離，防止上層線程池跑滿，影響二級降級邏輯調用.

 11:顯示調用fallback邏輯,用於特殊業務處理

 

public class CommandFacadeWithPrimarySecondary extends HystrixCommand<String> {
    private final static DynamicBooleanProperty usePrimary = DynamicPropertyFactory.getInstance().getBooleanProperty("primarySecondary.usePrimary", true);
    private final int id;
    public CommandFacadeWithPrimarySecondary(int id) {
        super(Setter
                .withGroupKey(HystrixCommandGroupKey.Factory.asKey("SystemX"))
                .andCommandKey(HystrixCommandKey.Factory.asKey("PrimarySecondaryCommand"))
                .andCommandPropertiesDefaults(
                        HystrixCommandProperties.Setter()
                                .withExecutionIsolationStrategy(ExecutionIsolationStrategy.SEMAPHORE)));
        this.id = id;
    }
    @Override
    protected String run() {
        if (usePrimary.get()) {
            return new PrimaryCommand(id).execute();
        } else {
            return new SecondaryCommand(id).execute();
        }
    }
    @Override
    protected String getFallback() {
        return "static-fallback-" + id;
    }
    @Override
    protected String getCacheKey() {
        return String.valueOf(id);
    }
    private static class PrimaryCommand extends HystrixCommand<String> {
        private final int id;
        private PrimaryCommand(int id) {
            super(Setter
                    .withGroupKey(HystrixCommandGroupKey.Factory.asKey("SystemX"))
                    .andCommandKey(HystrixCommandKey.Factory.asKey("PrimaryCommand"))
                    .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("PrimaryCommand"))
                    .andCommandPropertiesDefaults(
                            // we default to a 600ms timeout for primary
                            HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(600)));
            this.id = id;
        }
        @Override
        protected String run() {
            // perform expensive 'primary' service call
            return "responseFromPrimary-" + id;
        }
    }
    private static class SecondaryCommand extends HystrixCommand<String> {
        private final int id;
        private SecondaryCommand(int id) {
            super(Setter
                    .withGroupKey(HystrixCommandGroupKey.Factory.asKey("SystemX"))
                    .andCommandKey(HystrixCommandKey.Factory.asKey("SecondaryCommand"))
                    .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("SecondaryCommand"))
                    .andCommandPropertiesDefaults(
                            // we default to a 100ms timeout for secondary
                            HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(100)));
            this.id = id;
        }
        @Override
        protected String run() {
            // perform fast 'secondary' service call
            return "responseFromSecondary-" + id;
        }
    }
    public static class UnitTest {
        @Test
        public void testPrimary() {
            HystrixRequestContext context = HystrixRequestContext.initializeContext();
            try {
                ConfigurationManager.getConfigInstance().setProperty("primarySecondary.usePrimary", true);
                assertEquals("responseFromPrimary-20", new CommandFacadeWithPrimarySecondary(20).execute());
            } finally {
                context.shutdown();
                ConfigurationManager.getConfigInstance().clear();
            }
        }
        @Test
        public void testSecondary() {
            HystrixRequestContext context = HystrixRequestContext.initializeContext();
            try {
                ConfigurationManager.getConfigInstance().setProperty("primarySecondary.usePrimary", false);
                assertEquals("responseFromSecondary-20", new CommandFacadeWithPrimarySecondary(20).execute());
            } finally {
                context.shutdown();
                ConfigurationManager.getConfigInstance().clear();
            }
        }
    }
}

 

 NOTE:顯示調用降級適用於特殊需求的場景,fallback用於業務處理，fallback不再承擔降級職責，建議慎重使用，會造成監控統計換亂等問題.

12:命令調用合並:HystrixCollapser

命令調用合並允許多個請求合並到一個線程/信號下批量執行。

執行流程圖如下:

public class CommandCollapserGetValueForKey extends HystrixCollapser<List<String>, String, Integer> {
    private final Integer key;
    public CommandCollapserGetValueForKey(Integer key) {
        this.key = key;
    }
    @Override
    public Integer getRequestArgument() {
        return key;
    }
    @Override
    protected HystrixCommand<List<String>> createCommand(final Collection<CollapsedRequest<String, Integer>> requests) {
        //創建返回command對象
        return new BatchCommand(requests);
    }
    @Override
    protected void mapResponseToRequests(List<String> batchResponse, Collection<CollapsedRequest<String, Integer>> requests) {
        int count = 0;
        for (CollapsedRequest<String, Integer> request : requests) {
            //手動匹配請求和響應
            request.setResponse(batchResponse.get(count++));
        }
    }
    private static final class BatchCommand extends HystrixCommand<List<String>> {
        private final Collection<CollapsedRequest<String, Integer>> requests;
        private BatchCommand(Collection<CollapsedRequest<String, Integer>> requests) {
                super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))
                    .andCommandKey(HystrixCommandKey.Factory.asKey("GetValueForKey")));
            this.requests = requests;
        }
        @Override
        protected List<String> run() {
            ArrayList<String> response = new ArrayList<String>();
            for (CollapsedRequest<String, Integer> request : requests) {
                response.add("ValueForKey: " + request.getArgument());
            }
            return response;
        }
    }
    public static class UnitTest {
        HystrixRequestContext context = HystrixRequestContext.initializeContext();
        try {
            Future<String> f1 = new CommandCollapserGetValueForKey(1).queue();
            Future<String> f2 = new CommandCollapserGetValueForKey(2).queue();
            Future<String> f3 = new CommandCollapserGetValueForKey(3).queue();
            Future<String> f4 = new CommandCollapserGetValueForKey(4).queue();
            assertEquals("ValueForKey: 1", f1.get());
            assertEquals("ValueForKey: 2", f2.get());
            assertEquals("ValueForKey: 3", f3.get());
            assertEquals("ValueForKey: 4", f4.get());
            assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
            HystrixCommand<?> command = HystrixRequestLog.getCurrentRequest().getExecutedCommands().toArray(new HystrixCommand<?>[1])[0];
            assertEquals("GetValueForKey", command.getCommandKey().name());
            assertTrue(command.getExecutionEvents().contains(HystrixEventType.COLLAPSED));
            assertTrue(command.getExecutionEvents().contains(HystrixEventType.SUCCESS));
        } finally {
         context.shutdown();
        }   
    }
}

 

 NOTE:使用場景:HystrixCollapser用於對多個相同業務的請求合並到一個線程甚至可以合並到一個連接中執行，降低線程交互次和IO數,但必須保證他們屬於同一依賴.

四:監控平台搭建Hystrix-dashboard

1:監控dashboard介紹

dashboard面板可以對依賴關鍵指標提供實時監控,如下圖:

 

2:實例暴露command統計數據

Hystrix使用Servlet對當前JVM下所有command調用情況作數據流輸出

配置如下:

 

<servlet>
    <display-name>HystrixMetricsStreamServlet</display-name>
    <servlet-name>HystrixMetricsStreamServlet</servlet-name>
    <servlet-class>com.netflix.hystrix.contrib.metrics.eventstream.HystrixMetricsStreamServlet</servlet-class>
</servlet>
<servlet-mapping>
    <servlet-name>HystrixMetricsStreamServlet</servlet-name>
    <url-pattern>/hystrix.stream</url-pattern>
</servlet-mapping>
<!-- 
    對應URL格式 : http://hostname:port/application/hystrix.stream
-->

 

3:集群模式監控統計搭建

1)使用Turbine組件做集群數據匯總

結構圖如下;

2)內嵌jetty提供Servlet容器,暴露HystrixMetrics

public class JettyServer {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    private int port;
    private ExecutorService executorService = Executors.newFixedThreadPool(1);
    private Server server = null;
    public void init() {
        try {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        //綁定8080端口,加載HystrixMetricsStreamServlet並映射url
                        server = new Server(8080);
                        WebAppContext context = new WebAppContext();
                        context.setContextPath("/");
                        context.addServlet(HystrixMetricsStreamServlet.class, "/hystrix.stream");
                        context.setResourceBase(".");
                        server.setHandler(context);
                        server.start();
                        server.join();
                    } catch (Exception e) {
                        logger.error(e.getMessage(), e);
                    }
                }
            });
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }
    public void destory() {
        if (server != null) {
            try {
                server.stop();
                server.destroy();
                logger.warn("jettyServer stop and destroy!");
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        }
    }
}

 

3)Turbine搭建和配置

  a:配置Turbine Servlet收集器

<servlet>
   <description></description>
   <display-name>TurbineStreamServlet</display-name>
   <servlet-name>TurbineStreamServlet</servlet-name>
   <servlet-class>com.netflix.turbine.streaming.servlet.TurbineStreamServlet</servlet-class>
 </servlet>
 <servlet-mapping>
   <servlet-name>TurbineStreamServlet</servlet-name>
   <url-pattern>/turbine.stream</url-pattern>
 </servlet-mapping>

 

   b:編寫config.properties配置集群實例

#配置兩個集群:mobil-online,ugc-online
turbine.aggregator.clusterConfig=mobil-online,ugc-online
#配置mobil-online集群實例
turbine.ConfigPropertyBasedDiscovery.mobil-online.instances=10.10.*.*,10.10.*.*,10.10.*.*,10.10.*.*,10.10.*.*,10.10.*.*,10.16.*.*,10.16.*.*,10.16.*.*,10.16.*.*
#配置mobil-online數據流servlet
turbine.instanceUrlSuffix.mobil-online=:8080/hystrix.stream
#配置ugc-online集群實例
turbine.ConfigPropertyBasedDiscovery.ugc-online.instances=10.10.*.*,10.10.*.*,10.10.*.*,10.10.*.*#配置ugc-online數據流servlet
turbine.instanceUrlSuffix.ugc-online=:8080/hystrix.stream

 

  c:使用Dashboard配置連接Turbine

  如下圖 :

 

五:Hystrix配置與分析

1:Hystrix 配置

1):Command 配置

Command配置源碼在HystrixCommandProperties,構造Command時通過Setter進行配置

具體配置解釋和默認值如下

//使用命令調用隔離方式,默認:采用線程隔離,ExecutionIsolationStrategy.THREAD
private final HystrixProperty<ExecutionIsolationStrategy> executionIsolationStrategy; 
//使用線程隔離時，調用超時時間，默認:1秒
private final HystrixProperty<Integer> executionIsolationThreadTimeoutInMilliseconds; 
//線程池的key,用於決定命令在哪個線程池執行
private final HystrixProperty<String> executionIsolationThreadPoolKeyOverride; 
//使用信號量隔離時，命令調用最大的並發數,默認:10
private final HystrixProperty<Integer> executionIsolationSemaphoreMaxConcurrentRequests;
//使用信號量隔離時，命令fallback(降級)調用最大的並發數,默認:10
private final HystrixProperty<Integer> fallbackIsolationSemaphoreMaxConcurrentRequests; 
//是否開啟fallback降級策略 默認:true 
private final HystrixProperty<Boolean> fallbackEnabled; 
// 使用線程隔離時，是否對命令執行超時的線程調用中斷（Thread.interrupt()）操作.默認:true
private final HystrixProperty<Boolean> executionIsolationThreadInterruptOnTimeout; 
// 統計滾動的時間窗口,默認:5000毫秒circuitBreakerSleepWindowInMilliseconds
private final HystrixProperty<Integer> metricsRollingStatisticalWindowInMilliseconds;
// 統計窗口的Buckets的數量,默認:10個,每秒一個Buckets統計
private final HystrixProperty<Integer> metricsRollingStatisticalWindowBuckets; // number of buckets in the statisticalWindow
//是否開啟監控統計功能,默認:true
private final HystrixProperty<Boolean> metricsRollingPercentileEnabled; 
// 是否開啟請求日志,默認:true
private final HystrixProperty<Boolean> requestLogEnabled; 
//是否開啟請求緩存,默認:true
private final HystrixProperty<Boolean> requestCacheEnabled; // Whether request caching is enabled.

 

2):熔斷器（Circuit Breaker）配置

Circuit Breaker配置源碼在HystrixCommandProperties,構造Command時通過Setter進行配置,每種依賴使用一個Circuit Breaker

// 熔斷器在整個統計時間內是否開啟的閥值，默認20秒。也就是10秒鍾內至少請求20次，熔斷器才發揮起作用  
private final HystrixProperty<Integer> circuitBreakerRequestVolumeThreshold;   
//熔斷器默認工作時間,默認:5秒.熔斷器中斷請求5秒后會進入半打開狀態,放部分流量過去重試  
private final HystrixProperty<Integer> circuitBreakerSleepWindowInMilliseconds;   
//是否啟用熔斷器,默認true. 啟動  
private final HystrixProperty<Boolean> circuitBreakerEnabled;   
//默認:50%。當出錯率超過50%后熔斷器啟動.  
private final HystrixProperty<Integer> circuitBreakerErrorThresholdPercentage;  
//是否強制開啟熔斷器阻斷所有請求,默認:false,不開啟  
private final HystrixProperty<Boolean> circuitBreakerForceOpen;   
//是否允許熔斷器忽略錯誤,默認false, 不開啟  
private final HystrixProperty<Boolean> circuitBreakerForceClosed; 

 

 

3):命令合並(Collapser)配置

Command配置源碼在HystrixCollapserProperties,構造Collapser時通過Setter進行配置

//請求合並是允許的最大請求數,默認: Integer.MAX_VALUE  
private final HystrixProperty<Integer> maxRequestsInBatch;  
//批處理過程中每個命令延遲的時間,默認:10毫秒  
private final HystrixProperty<Integer> timerDelayInMilliseconds;  
//批處理過程中是否開啟請求緩存,默認:開啟  
private final HystrixProperty<Boolean> requestCacheEnabled;

 

 

 

4):線程池(ThreadPool)配置

/** 
配置線程池大小,默認值10個. 
建議值:請求高峰時99.5%的平均響應時間 + 向上預留一些即可 
*/  
HystrixThreadPoolProperties.Setter().withCoreSize(int value)  
/** 
配置線程值等待隊列長度,默認值:-1 
建議值:-1表示不等待直接拒絕,測試表明線程池使用直接決絕策略+ 合適大小的非回縮線程池效率最高.所以不建議修改此值。 
當使用非回縮線程池時，queueSizeRejectionThreshold,keepAliveTimeMinutes 參數無效 
*/  
HystrixThreadPoolProperties.Setter().withMaxQueueSize(int value) 

 

 

2:Hystrix關鍵組件分析

 1):Hystrix流程結構解析

流程說明:

1:每次調用創建一個新的HystrixCommand,把依賴調用封裝在run()方法中.

2:執行execute()/queue做同步或異步調用.

3:判斷熔斷器(circuit-breaker)是否打開,如果打開跳到步驟8,進行降級策略,如果關閉進入步驟.

4:判斷線程池/隊列/信號量是否跑滿，如果跑滿進入降級步驟8,否則繼續后續步驟.

5:調用HystrixCommand的run方法.運行依賴邏輯

5a:依賴邏輯調用超時,進入步驟8.

6:判斷邏輯是否調用成功

6a:返回成功調用結果

6b:調用出錯，進入步驟8.

7:計算熔斷器狀態,所有的運行狀態(成功, 失敗, 拒絕,超時)上報給熔斷器，用於統計從而判斷熔斷器狀態.

8:getFallback()降級邏輯.

  以下四種情況將觸發getFallback調用：

 (1):run()方法拋出非HystrixBadRequestException異常。

 (2):run()方法調用超時

 (3):熔斷器開啟攔截調用

 (4):線程池/隊列/信號量是否跑滿

8a:沒有實現getFallback的Command將直接拋出異常

8b:fallback降級邏輯調用成功直接返回

8c:降級邏輯調用失敗拋出異常

9:返回執行成功結果

2):熔斷器:Circuit Breaker 

Circuit Breaker 流程架構和統計

每個熔斷器默認維護10個bucket,每秒一個bucket,每個blucket記錄成功,失敗,超時,拒絕的狀態，

默認錯誤超過50%且10秒內超過20個請求進行中斷攔截. 

3)隔離(Isolation)分析

Hystrix隔離方式采用線程/信號的方式,通過隔離限制依賴的並發量和阻塞擴散.

(1):線程隔離

         把執行依賴代碼的線程與請求線程(如:jetty線程)分離，請求線程可以自由控制離開的時間(異步過程)。

   通過線程池大小可以控制並發量，當線程池飽和時可以提前拒絕服務,防止依賴問題擴散。

   線上建議線程池不要設置過大，否則大量堵塞線程有可能會拖慢服務器。

(2):線程隔離的優缺點

線程隔離的優點:

[1]:使用線程可以完全隔離第三方代碼,請求線程可以快速放回。

[2]:當一個失敗的依賴再次變成可用時，線程池將清理，並立即恢復可用，而不是一個長時間的恢復。

[3]:可以完全模擬異步調用，方便異步編程。

線程隔離的缺點:

[1]:線程池的主要缺點是它增加了cpu，因為每個命令的執行涉及到排隊(默認使用SynchronousQueue避免排隊)，調度和上下文切換。

[2]:對使用ThreadLocal等依賴線程狀態的代碼增加復雜性，需要手動傳遞和清理線程狀態。

NOTE: Netflix公司內部認為線程隔離開銷足夠小，不會造成重大的成本或性能的影響。

Netflix 內部API 每天100億的HystrixCommand依賴請求使用線程隔，每個應用大約40多個線程池，每個線程池大約5-20個線程。

(3):信號隔離

      信號隔離也可以用於限制並發訪問，防止阻塞擴散, 與線程隔離最大不同在於執行依賴代碼的線程依然是請求線程（該線程需要通過信號申請）,

   如果客戶端是可信的且可以快速返回，可以使用信號隔離替換線程隔離,降低開銷.

   

線程隔離與信號隔離區別如下圖:

 解析圖片出自官網wiki , 更多內容請見官網: https://github.com/Netflix/Hystrix