Hystrix入門指南

Introduction

 

1、Where does the name come from?

hystrix對應的中文名字是“豪豬”，豪豬周身長滿了刺，能保護自己不受天敵的傷害，代表了一種防御機制，這與hystrix本身的功能不謀而合，因此Netflix團隊將該框架命名為Hystrix，並使用了對應的卡通形象做作為logo。

2、What Is Hystrix?

在一個分布式系統里，許多依賴不可避免的會調用失敗，比如超時、異常等，如何能夠保證在一個依賴出問題的情況下，不會導致整體服務失敗，這個就是Hystrix需要做的事情。Hystrix提供了熔斷、隔離、Fallback、cache、監控等功能，能夠在一個、或多個依賴同時出現問題時保證系統依然可用。

 

   

3、Hello Hystrix

public class CommandHelloWorld extends HystrixCommand<String> {
 
    private final String name;
 
    public CommandHelloWorld(String name) {
        super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup")); //必須
        this.name = name;
    }
 
    @Override
    protected String run() {
        /*
         網絡調用 或者其他一些業務邏輯，可能會超時或者拋異常
        */
        return "Hello " + name + "!";
    }
}
  
String s = new CommandHelloWorld("Bob").execute(); //
Future<String> s = new CommandHelloWorld("Bob").queue();
Observable<String> s = new CommandHelloWorld("Bob").observe();
Observable<String> s = new CommandHelloWorld("Bob").toObservable()

    說明：

• execute() — blocks, then returns the single response received from the dependency (or throws an exception in case of an error)
• queue() — returns a Future with which you can obtain the single response from the dependency
• observe() — subscribes to the Observable that represents the response(s) from the dependency and returns an Observable that replicates that source Observable
• toObservable() — returns an Observable that, when you subscribe to it, will execute the Hystrix command and emit its responses

 

4、Flow Chart

 

 說明：

1. Construct a HystrixCommand or HystrixObservableCommand Object
2. Execute the Command
3. Is the Response Cached?
4. Is the Circuit Open?
5. Is the Thread Pool/Queue/Semaphore Full?
6. HystrixObservableCommand.construct() or HystrixCommand.run()
7. Calculate Circuit Health
8. Get the Fallback
9. Return the Successful Response

 常用功能介紹

依賴隔離

一個用戶請求的成功執行，肯能依賴數十上百個外部服務，如果沒有隔離，單個依賴的失敗，可能會印象其他依賴的正常執行。如下圖所示，為每個依賴配置了單獨線程池

   

在下圖中，當Dep I 出現問題時，DepA 和Dep M大以來可以正常執行

 線程池隔離的使用例子

public class CommandHelloWorld extends HystrixCommand<String> {
 
    private final String name;
 
    public CommandHelloWorld(String name) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))  //必須
                .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("ExampleGroup-pool"))  //可選,默認 使用 this.getClass().getSimpleName();
                .andThreadPoolPropertiesDefaults(HystrixThreadPoolProperties.Setter().withCoreSize(4)));
         
        this.name = name;
    }
 
    @Override
    protected String run() throws InterruptedException {
        System.out.println("running");
        TimeUnit.MILLISECONDS.sleep(1000);
        return "Hello " + name + "!";
    }
     
}

線程池常用參數設置：

實現類：HystrixThreadPoolProperties

名稱	類型	含義	默認值
coreSize	Integer	線程池大小	10
maxQueueSize	Integer	隊列大小，一經初始化后不能修改	-1
queueSizeRejectionThreshold	Integer	隊列reject閾值，可以動態修改 maxQueueSize>0是生效，一般設置為小於 maxQueueSizede 的數值	5

 

 

 

 

 

 

 

 

 

Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))  //必須
 .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
                .withExecutionTimeoutInMilliseconds(500))  //超時時間
 .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("ExampleGroup-pool"))  //可選,默認 使用 this.getClass().getSimpleName();
 .andThreadPoolPropertiesDefaults(HystrixThreadPoolProperties.Setter().withCoreSize(4)
        .withMaxQueueSize(10).withQueueSizeRejectionThreshold(7))

 

Q： 怎么設置線程池大小？

  A：Qps* Tp99 +冗余線程

信號量隔離

線程池隔離中，發起請求的線程和真實執行的線程不是同一個線程，使用信號量隔離時，它們是同一個線程， 兩種隔離的區別如下圖：

public class CommandUsingSemaphoreIsolation extends HystrixCommand<String> {
 
    private final int id;
    private long start,end ;
 
    public CommandUsingSemaphoreIsolation(int id) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))
                // since we're doing an in-memory cache lookup we choose SEMAPHORE isolation
                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
                        .withExecutionIsolationStrategy(HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE) //設置使用信號量隔離策略
                                .withExecutionIsolationSemaphoreMaxConcurrentRequests(3)  //設置信號量隔離時的最大並發請求數
                                .withFallbackIsolationSemaphoreMaxConcurrentRequests(5)     //設置fallback的最大並發數
                        .withExecutionTimeoutInMilliseconds(300)));   //設置超時時間
        this.id = id;
        this.start = System.currentTimeMillis();
    }
 
    @Override
    protected String run() throws InterruptedException {
        // a real implementation would retrieve data from in memory data structure
        TimeUnit.MILLISECONDS.sleep(id*30);
        System.out.println("running normal, id="+id);
        return "ValueFromHashMap_" + id;
    }
 
    @Override
    protected String getFallback(){
        System.out.println(" fallback, id="+id);
        return "fallback:"+id;
    }
 
}
  
@Test
public void maxCurrentRequst() throws InterruptedException {
    int count =10;
    while (count >0){
        int id = count--;
        new Thread(() -> {
            try {
                new CommandUsingSemaphoreIsolation(id).execute();
            }catch (Exception ex){
                System.out.println("Exception:"+ex.getMessage()+" id="+id);
            }
        }).start();
    }
 
    TimeUnit.SECONDS.sleep(100);
}
 //注：使用信號量隔離，在同一個線程中即使循環調用new CommandUsingSemaphoreIsolation(id).queue()，run方法也是順序執行;

//控制台輸出

fallback, id=10
fallback, id=9
fallback, id=5
fallback, id=8
fallback, id=1
Exception:CommandUsingSemaphoreIsolation fallback execution rejected. id=4
Exception:CommandUsingSemaphoreIsolation fallback execution rejected. id=7
running normal, id=2
running normal, id=3
running normal, id=6

Q: 什么時候使用線程池隔離，什么使用信號量隔離？

A:  線程池隔離缺點是帶來一定的開銷，但不會阻塞請求線程，適合於於IO密集型的任務

信號量隔離使用用戶請求線程，沒有格外線程切換開銷，使用與執行時間和執行邏輯都比較短的本地計算。比如CPU密集型的任務

Fallback

Q1: 為什么需要fallback?

簡單來說，在依賴調用失敗時，我們一般會需要提供降級方案，Hystrix對此提供了支持

降級  

public class CommandHelloWorld extends HystrixCommand<String> {
 
    private final String name;
 
    public CommandHelloWorld(String name) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))  //必須
                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
                        .withExecutionTimeoutInMilliseconds(500))  //超時時間
                .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("ExampleGroup-pool"))  //可選,默認 使用 this.getClass().getSimpleName();
                .andThreadPoolPropertiesDefaults(HystrixThreadPoolProperties.Setter().withCoreSize(4)));
 
        this.name = name;
    }
 
    @Override
    protected String run() throws InterruptedException {
        System.out.println("running");
        TimeUnit.MILLISECONDS.sleep(1000);
        return "Hello " + name + "!";
    }
 
    @Override
    protected String getFallback() {
        return "Hello "+"Fallback";
    }
}
  
@Test
public void fallbackTest(){
    assertEquals("Hello Fallback",new CommandHelloWorld("World").execute());
}

 Q2：什么情況下會觸發fallback？

簡單來說，就是run方法拋異常，超時，線程/信號量reject、短路

Failure Type	Exception class	Exception.cause	subject to fallback
FAILURE	HystrixRuntimeException	underlying exception (user-controlled)	YES
TIMEOUT	HystrixRuntimeException	j.u.c.TimeoutException	YES
SHORT_CIRCUITED	HystrixRuntimeException	j.l.RuntimeException	YES
THREAD_POOL_REJECTED	HystrixRuntimeException	j.u.c.RejectedExecutionException	YES
SEMAPHORE_REJECTED	HystrixRuntimeException	j.l.RuntimeException	YES
BAD_REQUEST	HystrixBadRequestException	underlying exception (user-controlled)	NO

 

 

 

 

 

 

以下為測試的主程序：

public class CommandHelloFailure extends HystrixCommand<String> {
 
    private final String name;
 
    public CommandHelloFailure(String name) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))  //必須
                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
                        .withExecutionTimeoutInMilliseconds(1000))  //超時時間
                .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("ExampleGroup-pool"))
                .andThreadPoolPropertiesDefaults(HystrixThreadPoolProperties.Setter().withCoreSize(3)));
 
        this.name = name;
    }
 
    @Override
    protected String run() throws InterruptedException {
        String theadName = this.getThreadPoolKey().name();
        String cmdKey=this.getThreadPoolKey().name();
        System.out.println("running begin , threadPool="+theadName+" cmdKey="+cmdKey+" name="+name);
 
        if("Exception".equals(name)) {
            throw new RuntimeException("this command always fails");
        }else if("Timeout".equals(name)){
            TimeUnit.SECONDS.sleep(2);
        }else if("Reject".equals(name)){
            TimeUnit.MILLISECONDS.sleep(800);
        }
        System.out.println(" run end");
 
        return "Hello " + name + "!";
    }
 
    @Override
    protected String getFallback() {
        StringBuilder sb = new StringBuilder("running fallback");
        boolean isRejected = isResponseRejected();
        boolean isException = isFailedExecution();
        boolean isTimeout= isResponseTimedOut();
        boolean isCircut = isCircuitBreakerOpen();
 
        sb.append(", isRejected:").append(isRejected);
        sb.append(", isException:"+isException);
        if(isException){
            sb.append(" msg=").append(getExecutionException().getMessage());
        }
        sb.append(",  isTimeout: "+isTimeout);
        sb.append(",  isCircut:"+isCircut);
 
        sb.append(", group:").append(this.getCommandGroup().name());
        sb.append(", threadpool:").append(getThreadPoolKey().name());
        System.out.println(sb.toString());
 
        String msg="Hello Failure " + name + "!";
        return msg;
    }
}

 

FAILURE 

測試由異常導致的fallback

@Test
public void expTest() {
    assertEquals("Hello Failure Exception!", new CommandHelloFailure("Exception").execute());
}
  

//控制台輸出

running begin , threadPool=ExampleGroup-pool cmdKey=ExampleGroup-pool name=Exception
running fallback, isRejected:false, isException:true msg=this command always fails, isTimeout: false, isCircut:false, group:ExampleGroup, threadpool:ExampleGroup-pool

 

TIMEOUT

測試有超時導致的fallback

 

@Test
public void timeOutTest() {
    assertEquals("Hello Failure Timeout!", new CommandHelloFailure("Timeout").execute());
}
  

//控制台輸出

running begin , threadPool=ExampleGroup-pool cmdKey=ExampleGroup-pool name=Timeout
running fallback, isRejected:false, isException:false, isTimeout: true, isCircut:false, group:ExampleGroup, threadpool:ExampleGroup-pool

 

THREAD_POOL_REJECTED

並發執行的任務數超過線程池和隊列之和會被reject，導致fallback

@Test
public void rejectTest() throws InterruptedException {
    int count = 5;
    while (count-- > 0){
        new CommandHelloFailure("Reject").queue();
        TimeUnit.MILLISECONDS.sleep(100);
    }
}

//控制台輸出

running begin , threadPool=ExampleGroup-pool cmdKey=ExampleGroup-pool name=Reject
running begin , threadPool=ExampleGroup-pool cmdKey=ExampleGroup-pool name=Reject
running begin , threadPool=ExampleGroup-pool cmdKey=ExampleGroup-pool name=Reject
running fallback, isRejected:true, isException:false, isTimeout: false, isCircut:false, group:ExampleGroup, threadpool:ExampleGroup-pool
running fallback, isRejected:true, isException:false, isTimeout: false, isCircut:false, group:ExampleGroup, threadpool:ExampleGroup-pool

SEMAPHORE_REJECTED  與 THREAD_POOL_REJECTED 類似，不再演示

 

SHORT_CIRCUITED

在一定時間內，用戶請求超過一定的比例失敗時（timeout, failure, reject），斷路器就會打開；短路器打開后所有請求直接走fallback

參數設置

名稱	類型	含義	默認值
circuitBreakerEnabled	Boolean	是否啟用斷路器	true
circuitBreakerErrorThresholdPercentage	Integer	錯誤百分比，超過該值打開斷路器	50
circuitBreakerForceClosed	Boolean	強制斷路器打開	false
circuitBreakerForceOpen	Boolean	強制短路器關閉	false
circuitBreakerRequestVolumeThreshold	Integer	10s中內最少的請求量，大於該值，斷路器配置才會生效	20
circuitBreakerSleepWindowInMilliseconds	Integer	短路器打開后多長時間嘗試關閉（Half open）	5s

 

 

 

 

 

 

 

一般配置如下：

Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))  //必須
        .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
                .withExecutionTimeoutInMilliseconds(50)//超時時間
                .withCircuitBreakerRequestVolumeThreshold(5)
                .withCircuitBreakerSleepWindowInMilliseconds(1000)
                .withCircuitBreakerErrorThresholdPercentage(50))
        .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("ExampleGroup-pool"))  //可選,默認 使用 this.getClass().getSimpleName();
        .andThreadPoolPropertiesDefaults(HystrixThreadPoolProperties.Setter().withCoreSize(4));

以上配置的含義是: 在10s內，如果請求在5個及以上，且有50%失敗的情況下，開啟斷路器；斷路器開啟1000ms后嘗試關閉

短路器的工作機制，引用自官方文檔：

The precise way that the circuit opening and closing occurs is as follows:
Assuming the volume across a circuit meets a certain threshold (HystrixCommandProperties.circuitBreakerRequestVolumeThreshold())...
And assuming that the error percentage exceeds the threshold error percentage (HystrixCommandProperties.circuitBreakerErrorThresholdPercentage())...
Then the circuit-breaker transitions from CLOSED to OPEN.
While it is open, it short-circuits all requests made against that circuit-breaker.
After some amount of time (HystrixCommandProperties.circuitBreakerSleepWindowInMilliseconds()), the next single request is let through (this is the HALF-OPEN state). If the request fails, the circuit-breaker returns to the OPEN state for the duration of the sleep window. If the request succeeds, the circuit-breaker transitions to CLOSED and the logic in 1. takes over again.

Q3：fallback時我們應該怎么辦？

一般有以下幾種策略：

1、不實現getFallback方法：依賴調用失敗時直接拋出異常

2、實現getFallback方法，返回默認值：這是一種常見的策略

3、實現getFallback方法，走降級方案

此外，生產環境中，fallback時，一般需要打點記錄

請求合並

簡單來說，就是將一段時間內的多次請求合並為一次請求，常用於網絡IO中，能減少IO次數，缺點是增加平均延遲

以下是測試代碼主程序：

public class CommandCollapserGetValueForKey extends HystrixCollapser<List<String>, String, Integer> {
 
    private final Integer key;
 
    public CommandCollapserGetValueForKey(Integer key) {
        super(Setter.withCollapserKey(HystrixCollapserKey.Factory.asKey("Collapser"))
                .andCollapserPropertiesDefaults(HystrixCollapserProperties.Setter()
                        .withMaxRequestsInBatch(3)
                .withTimerDelayInMilliseconds(10)));
        this.key = key;
    }
 
    @Override
    public Integer getRequestArgument() {
        return key;
    }
 
    @Override
    protected HystrixCommand<List<String>> createCommand(final Collection<CollapsedRequest<String, Integer>> requests) {
        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() {
            System.out.println("BatchCommand run  "+requests.size());
            ArrayList<String> response = new ArrayList<String>();
            for (CollapsedRequest<String, Integer> request : requests) {
                // artificial response for each argument received in the batch
                response.add("ValueForKey: " + request.getArgument());
            }
            return response;
        }
    }
}
  
  
@Test
public void testCollapser() throws Exception {
    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());
 
        // assert that the batch command 'GetValueForKey' was in fact
        // executed and that it executed only once
        assertEquals(2, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
        HystrixCommand<?> command = HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().toArray(new HystrixCommand<?>[1])[0];
        // assert the command is the one we're expecting
        assertEquals("GetValueForKey", command.getCommandKey().name());
        // confirm that it was a COLLAPSED command execution
        assertTrue(command.getExecutionEvents().contains(HystrixEventType.COLLAPSED));
        // and that it was successful
        assertTrue(command.getExecutionEvents().contains(HystrixEventType.SUCCESS));
    } finally {
        context.shutdown();
    }
}
  
//控制輸出
BatchCommand run  3
BatchCommand run  1

執行流程：

 

使用該特性

1、必須繼承HystrixCollapser類，

2、實現以下方法：

getRequestArgument： 返回請求參數對象

createCommand ： 返回BatchCommand

mapResponseToRequests：實現Response和Request的映射

3、創建對應的BatchCommand類：批量請求的具體實現

 

參數配置：

名稱	類型	含義	默認值
maxRequestsInBatch	Integer	每個批次最大的請求數，超過該值，創建新的batch請求	Integer.MAX_VALUE
timerDelayInMilliseconds	Integer	等待時間窗口，超過該值，創建新的batch請求	10ms
requestCacheEnabled	Boolean	是否啟用cache	true

 

 

 

 

 

 

一般配置如下

Setter.withCollapserKey(HystrixCollapserKey.Factory.asKey("Collapser"))
       .andCollapserPropertiesDefaults(HystrixCollapserProperties.Setter()
               .withMaxRequestsInBatch(3)
       .withTimerDelayInMilliseconds(5))；

 

請求cache

public class CommandUsingRequestCache extends HystrixCommand<Boolean> {
    private final int value;
     
    public CommandUsingRequestCache(int value) {
        super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"));
        this.value = value;
    }
 
    @Override
    public Boolean run() {
        return value == 0 || value % 2 == 0;
    }
 
   //使用cache功能，必須實現該方法
    @Override
    public String getCacheKey() {
        return String.valueOf(value);
    }
}
  
@Test
public void testWithCacheHits() {
    HystrixRequestContext context = HystrixRequestContext.initializeContext();
    try {
        CommandUsingRequestCache command2a = new CommandUsingRequestCache(2);
        CommandUsingRequestCache command2b = new CommandUsingRequestCache(2);
 
        assertTrue(command2a.execute());
        //第一次請求，沒有cache
        assertFalse(command2a.isResponseFromCache());
 
        assertTrue(command2b.execute());
        // 第二次請求，從cache中拿的結果
        assertTrue(command2b.isResponseFromCache());
    } finally {
        context.shutdown();
    }
 
    context = HystrixRequestContext.initializeContext();
    try {
        CommandUsingRequestCache command3b = new CommandUsingRequestCache(2);
        assertTrue(command3b.execute());
        // this is a new request context so this
        //new了新的 request context后，之前的cache失效
        assertFalse(command3b.isResponseFromCache());
    } finally {
        context.shutdown();
    }
}

Hystrix Context

 Global Context

UserRequest Context

 

使用與監控

1、工程中使用

使用Hystrix很簡單，只需要添加相應依賴即可，以Maven為例：

 

<!-- hystrix 依賴 -->
<dependency>
    <groupId>com.netflix.hystrix</groupId>
    <artifactId>hystrix-core</artifactId>
    <version>1.5.9</version>
</dependency>
<dependency>
    <groupId>com.netflix.hystrix</groupId>
    <artifactId>hystrix-metrics-event-stream</artifactId>
    <version>1.5.9</version>
</dependency>

2、DashBoard使用 

web.xml中配置相應的Servlet

<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>

下載附件中的war文件和jar文件到任意目錄，執行

java -jar jetty-runner-9.2.10.v20150310.jar --port 8410 hystrix-dashboard-1.5.1.war

然后在瀏覽器中打開：http://localhost:8410/  ，在輸入框中填寫 http://hostname:port/application/hystrix.stream， 點擊 Add Stream ,然后在點擊Monitor Stream， 看到如下圖：

每個指標對應的含義：

一般來說： Thread-pool Rejections  和Failuress/Exception應該是0，Thread timeouts是個很小的值。

代碼結構

附件

1、啟動腳本 start.sh

2、hystrix-dashboard-1.5.1.war

3、jetty-runner-9.2.10.v20150310.jar