Hystrix都停更了，我为什么还要学？-白红宇

最近小主看到很多公众号都在发布Hystrix停更的文章，spring cloud体系的使用者和拥护者一片哀嚎，实际上，spring作为Java最大的家族，根本不需要担心其中一两个零件的废弃，Hystrix的停更，只会催生更多或者更好的零件来替代它，因此，我们需要做的是：**知道Hystrix是干嘛，怎么用的，这样要找替代者就易于反掌了。

文章提纲：

为什么需要Hystrix?

Hystrix如何解决依赖隔离

如何使用Hystrix

1. 为什么需要Hystrix?

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

在高并发访问下,这些依赖的稳定性与否对系统的影响非常大,但是依赖有很多不可控问题:如网络连接缓慢，资源繁忙，暂时不可用，服务脱机等.

如下图：QPS为50的依赖 I 出现不可用，但是其他依赖仍然可用.

当依赖I 阻塞时,大多数服务器的线程池就出现阻塞(BLOCK),影响整个线上服务的稳定性.如下图:

在复杂的分布式架构的应用程序有很多的依赖，都会不可避免地在某些时候失败。高并发的依赖失败时如果没有隔离措施，当前应用服务就有被拖垮的风险。

1例如:一个依赖30个SOA服务的系统,每个服务99.99%可用。   299.99%的30次方 ≈ 99.7%   30.3% 意味着一亿次请求 会有 3,000,00次失败   4换算成时间大约每月有2个小时服务不稳定.   5随着服务依赖数量的变多，服务不稳定的概率会成指数性提高.   复制代码

解决问题方案:对依赖做隔离,Hystrix就是处理依赖隔离的框架,同时也是可以帮我们做依赖服务的治理和监控.

Netflix 公司开发并成功使用Hystrix,使用规模如下:

1he Netflix API processes 10+ billion HystrixCommand executions per day using thread isolation.    2Each API instance has 40+ thread-pools with 5-20 threads in each (most are set to 10).   3 复制代码

2. Hystrix如何解决依赖隔离

Hystrix使用命令模式HystrixCommand(Command)包装依赖调用逻辑，每个命令在单独线程中/信号授权下执行。

可配置依赖调用超时时间,超时时间一般设为比99.5%平均时间略高即可.当调用超时时，直接返回或执行fallback逻辑。

为每个依赖提供一个小的线程池（或信号），如果线程池已满调用将被立即拒绝，默认不采用排队.加速失败判定时间。

依赖调用结果分:成功，失败（抛出异常），超时，线程拒绝，短路。请求失败(异常，拒绝，超时，短路)时执行fallback(降级)逻辑。

提供熔断器组件,可以自动运行或手动调用,停止当前依赖一段时间(10秒)，熔断器默认错误率阈值为50%,超过将自动运行。

提供近实时依赖的统计和监控

Hystrix依赖的隔离架构,如下图:

3. 如何使用Hystrix

使用maven引入Hystrix依赖

1
         2
     
      1.3.16
         3
     
      1.1.2
          4  5
     
          6     
      
       com.netflix.hystrix
          7     
      
       hystrix-core
          8     
      
       ${hystrix.version}
          9 
        10     
     
         11     
      
       com.netflix.hystrix
         12     
      
       hystrix-metrics-event-stream
         13     
      
       ${hystrix-metrics-event-stream.version}
         14 
        15
        16
     
         17     
      
       nexus
         18     
      
       local private nexus
         19     
      
       http://maven.oschina.net/content/groups/public/
         20     
      
          21          
       
        true
          22     
         23     
      
          24          
       
        false
          25     
         26
        复制代码

使用命令模式封装依赖逻辑

1public class HelloWorldCommand extends HystrixCommand
     
       {    2    private final String name;    3    public HelloWorldCommand(String name) {    4        //最少配置:指定命令组名(CommandGroup)    5        super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"));    6        this.name = name;    7    }    8    @Override    9    protected String run() {   10        // 依赖逻辑封装在run()方法中   11        return "Hello " + name +" thread:" + Thread.currentThread().getName();   12    }   13    //调用实例   14    public static void main(String[] args) throws Exception{   15        //每个Command对象只能调用一次,不可以重复调用,   16        //重复调用对应异常信息:This instance can only be executed once. Please instantiate a new instance.   17        HelloWorldCommand helloWorldCommand = new HelloWorldCommand("Synchronous-hystrix");   18        //使用execute()同步调用代码,效果等同于:helloWorldCommand.queue().get();    19        String result = helloWorldCommand.execute();   20        System.out.println("result=" + result);   21 22        helloWorldCommand = new HelloWorldCommand("Asynchronous-hystrix");   23        //异步调用,可自由控制获取结果时机,   24        Future
      
        future = helloWorldCommand.queue();   25        //get操作不能超过command定义的超时时间,默认:1秒   26        result = future.get(100, TimeUnit.MILLISECONDS);   27        System.out.println("result=" + result);   28        System.out.println("mainThread=" + Thread.currentThread().getName());   29    }   30 31}   32    //运行结果: run()方法在不同的线程下执行   33    // result=Hello Synchronous-hystrix thread:hystrix-HelloWorldGroup-1   34    // result=Hello Asynchronous-hystrix thread:hystrix-HelloWorldGroup-2   35    // mainThread=main   复制代码

note:异步调用使用 command.queue()get(timeout, TimeUnit.MILLISECONDS);同步调用使用command.execute() 等同于 command.queue().get();

注册异步事件回调执行

1//注册观察者事件拦截    2Observable
     
       fs = new HelloWorldCommand("World").observe();    3//注册结果回调事件    4fs.subscribe(new Action1
      
       () {    5    @Override    6    public void call(String result) {    7         //执行结果处理,result 为HelloWorldCommand返回的结果    8        //用户对结果做二次处理.    9    }   10});   11//注册完整执行生命周期事件   12fs.subscribe(new Observer
       
        () {   13            @Override   14            public void onCompleted() {   15                // onNext/onError完成之后最后回调   16                System.out.println("execute onCompleted");   17            }   18            @Override   19            public void onError(Throwable e) {   20                // 当产生异常时回调   21                System.out.println("onError " + e.getMessage());   22                e.printStackTrace();   23            }   24            @Override   25            public void onNext(String v) {   26                // 获取结果后回调   27                System.out.println("onNext: " + v);   28            }   29        });   30/* 运行结果  31call execute result=Hello observe-hystrix thread:hystrix-HelloWorldGroup-3  32onNext: Hello observe-hystrix thread:hystrix-HelloWorldGroup-3  33execute onCompleted  34*/   复制代码

使用Fallback() 提供降级策略

1//重载HystrixCommand 的getFallback方法实现逻辑    2public class HelloWorldCommand extends HystrixCommand
     
       {    3    private final String name;    4    public HelloWorldCommand(String name) {    5        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("HelloWorldGroup"))    6                /* 配置依赖超时时间,500毫秒*/    7                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter().withExecutionIsolationThreadTimeoutInMilliseconds(500)));    8        this.name = name;    9    }   10    @Override   11    protected String getFallback() {   12        return "exeucute Falled";   13    }   14    @Override   15    protected String run() throws Exception {   16        //sleep 1 秒,调用会超时   17        TimeUnit.MILLISECONDS.sleep(1000);   18        return "Hello " + name +" thread:" + Thread.currentThread().getName();   19    }   20    public static void main(String[] args) throws Exception{   21        HelloWorldCommand command = new HelloWorldCommand("test-Fallback");   22        String result = command.execute();   23    }   24}   25/* 运行结果:getFallback() 调用运行  26getFallback executed  27*/   复制代码

NOTE: 除了HystrixBadRequestException异常之外，所有从run()方法抛出的异常都算作失败，并触发降级getFallback()和断路器逻辑。

1      HystrixBadRequestException用在非法参数或非系统故障异常等不应触发回退逻辑的场景。 复制代码

依赖命名:CommandKey

1public HelloWorldCommand(String name) {   2        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))   3                /* HystrixCommandKey工厂定义依赖名称 */   4                .andCommandKey(HystrixCommandKey.Factory.asKey("HelloWorld")));   5        this.name = name;   6    }   复制代码

NOTE: 每个CommandKey代表一个依赖抽象,相同的依赖要使用相同的CommandKey名称。依赖隔离的根本就是对相同CommandKey的依赖做隔离.

依赖分组:CommandGroup
命令分组用于对依赖操作分组,便于统计,汇总等.

1//使用HystrixCommandGroupKey工厂定义   2public HelloWorldCommand(String name) {   3    Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("HelloWorldGroup"))   4}   复制代码

NOTE: CommandGroup是每个命令最少配置的必选参数，在不指定ThreadPoolKey的情况下，字面值用于对不同依赖的线程池/信号区分.

线程池/信号:ThreadPoolKey

1public HelloWorldCommand(String name) {   2        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))   3                .andCommandKey(HystrixCommandKey.Factory.asKey("HelloWorld"))   4                /* 使用HystrixThreadPoolKey工厂定义线程池名称*/   5                .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("HelloWorldPool")));   6        this.name = name;   7    }   复制代码

NOTE: 当对同一业务依赖做隔离时使用CommandGroup做区分,但是对同一依赖的不同远程调用如(一个是redis 一个是http),可以使用HystrixThreadPoolKey做隔离区分.

1       最然在业务上都是相同的组，但是需要在资源上做隔离时，可以使用HystrixThreadPoolKey区分. 复制代码

请求缓存 Request-Cache

1public class RequestCacheCommand extends HystrixCommand
     
       {    2    private final int id;    3    public RequestCacheCommand( int id) {    4        super(HystrixCommandGroupKey.Factory.asKey("RequestCacheCommand"));    5        this.id = id;    6    }    7    @Override    8    protected String run() throws Exception {    9        System.out.println(Thread.currentThread().getName() + " execute id=" + id);   10        return "executed=" + id;   11    }   12    //重写getCacheKey方法,实现区分不同请求的逻辑   13    @Override   14    protected String getCacheKey() {   15        return String.valueOf(id);   16    }   17 18    public static void main(String[] args){   19        HystrixRequestContext context = HystrixRequestContext.initializeContext();   20        try {   21            RequestCacheCommand command2a = new RequestCacheCommand(2);   22            RequestCacheCommand command2b = new RequestCacheCommand(2);   23            Assert.assertTrue(command2a.execute());   24            //isResponseFromCache判定是否是在缓存中获取结果   25            Assert.assertFalse(command2a.isResponseFromCache());   26            Assert.assertTrue(command2b.execute());   27            Assert.assertTrue(command2b.isResponseFromCache());   28        } finally {   29            context.shutdown();   30        }   31        context = HystrixRequestContext.initializeContext();   32        try {   33            RequestCacheCommand command3b = new RequestCacheCommand(2);   34            Assert.assertTrue(command3b.execute());   35            Assert.assertFalse(command3b.isResponseFromCache());   36        } finally {   37            context.shutdown();   38        }   39    }   40}   复制代码

NOTE:请求缓存可以让(CommandKey/CommandGroup)相同的情况下,直接共享结果，降低依赖调用次数，在高并发和CacheKey碰撞率高场景下可以提升性能.

Servlet容器中，可以直接实用Filter机制Hystrix请求上下文

1public class HystrixRequestContextServletFilter implements Filter {    2    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)     3     throws IOException, ServletException {    4        HystrixRequestContext context = HystrixRequestContext.initializeContext();    5        try {    6            chain.doFilter(request, response);    7        } finally {    8            context.shutdown();    9        }   10    }   11}   12
     
         13      
      
       HystrixRequestContextServletFilter
         14      
      
       HystrixRequestContextServletFilter
         15      
      
       com.netflix.hystrix.contrib.requestservlet.HystrixRequestContextServletFilter
         16    
        17    
     
         18      
      
       HystrixRequestContextServletFilter
         19      
      
       /*
         20   
        21 复制代码

信号量隔离:SEMAPHORE
隔离本地代码或可快速返回远程调用(如memcached,redis)可以直接使用信号量隔离,降低线程隔离开销.

1public class HelloWorldCommand extends HystrixCommand
     
       {    2    private final String name;    3    public HelloWorldCommand(String name) {    4        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("HelloWorldGroup"))    5                /* 配置信号量隔离方式,默认采用线程池隔离 */    6                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter().withExecutionIsolationStrategy(HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE)));    7        this.name = name;    8    }    9    @Override   10    protected String run() throws Exception {   11        return "HystrixThread:" + Thread.currentThread().getName();   12    }   13    public static void main(String[] args) throws Exception{   14        HelloWorldCommand command = new HelloWorldCommand("semaphore");   15        String result = command.execute();   16        System.out.println(result);   17        System.out.println("MainThread:" + Thread.currentThread().getName());   18    }   19}   20/** 运行结果  21 HystrixThread:main  22 MainThread:main  23*/   复制代码

fallback降级逻辑命令嵌套

用场景:用于fallback逻辑涉及网络访问的情况,如缓存访问。

1public class CommandWithFallbackViaNetwork extends HystrixCommand
     
       {    2    private final int id;    3  4    protected CommandWithFallbackViaNetwork(int id) {    5        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("RemoteServiceX"))    6                .andCommandKey(HystrixCommandKey.Factory.asKey("GetValueCommand")));    7        this.id = id;    8    }    9 10    @Override   11    protected String run() {   12        // RemoteService.getValue(id);   13        throw new RuntimeException("force failure for example");   14    }   15 16    @Override   17    protected String getFallback() {   18        return new FallbackViaNetwork(id).execute();   19    }   20 21    private static class FallbackViaNetwork extends HystrixCommand
      
        {   22        private final int id;   23        public FallbackViaNetwork(int id) {   24            super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("RemoteServiceX"))   25                    .andCommandKey(HystrixCommandKey.Factory.asKey("GetValueFallbackCommand"))   26                    // 使用不同的线程池做隔离，防止上层线程池跑满，影响降级逻辑.   27                    .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("RemoteServiceXFallback")));   28            this.id = id;   29        }   30        @Override   31        protected String run() {   32            MemCacheClient.getValue(id);   33        }   34 35        @Override   36        protected String getFallback() {   37            return null;   38        }   39    }   40}   复制代码

NOTE:依赖调用和降级调用使用不同的线程池做隔离，防止上层线程池跑满，影响二级降级逻辑调用.

显示调用fallback逻辑,用于特殊业务处理

1public class CommandFacadeWithPrimarySecondary extends HystrixCommand
     
       {    2    private final static DynamicBooleanProperty usePrimary = DynamicPropertyFactory.getInstance().getBooleanProperty("primarySecondary.usePrimary", true);    3    private final int id;    4    public CommandFacadeWithPrimarySecondary(int id) {    5        super(Setter    6                .withGroupKey(HystrixCommandGroupKey.Factory.asKey("SystemX"))    7                .andCommandKey(HystrixCommandKey.Factory.asKey("PrimarySecondaryCommand"))    8                .andCommandPropertiesDefaults(    9                        HystrixCommandProperties.Setter()   10                                .withExecutionIsolationStrategy(ExecutionIsolationStrategy.SEMAPHORE)));   11        this.id = id;   12    }   13    @Override   14    protected String run() {   15        if (usePrimary.get()) {   16            return new PrimaryCommand(id).execute();   17        } else {   18            return new SecondaryCommand(id).execute();   19        }   20    }   21    @Override   22    protected String getFallback() {   23        return "static-fallback-" + id;   24    }   25    @Override   26    protected String getCacheKey() {   27        return String.valueOf(id);   28    }   29    private static class PrimaryCommand extends HystrixCommand
      
        {   30        private final int id;   31        private PrimaryCommand(int id) {   32            super(Setter   33                    .withGroupKey(HystrixCommandGroupKey.Factory.asKey("SystemX"))   34                    .andCommandKey(HystrixCommandKey.Factory.asKey("PrimaryCommand"))   35                    .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("PrimaryCommand"))   36                    .andCommandPropertiesDefaults(   37                            // we default to a 600ms timeout for primary   38                            HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(600)));   39            this.id = id;   40        }   41        @Override   42        protected String run() {   43            // perform expensive 'primary' service call   44            return "responseFromPrimary-" + id;   45        }   46    }   47    private static class SecondaryCommand extends HystrixCommand
       
         {   48        private final int id;   49        private SecondaryCommand(int id) {   50            super(Setter   51                    .withGroupKey(HystrixCommandGroupKey.Factory.asKey("SystemX"))   52                    .andCommandKey(HystrixCommandKey.Factory.asKey("SecondaryCommand"))   53                    .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("SecondaryCommand"))   54                    .andCommandPropertiesDefaults(   55                            // we default to a 100ms timeout for secondary   56                            HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(100)));   57            this.id = id;   58        }   59        @Override   60        protected String run() {   61            // perform fast 'secondary' service call   62            return "responseFromSecondary-" + id;   63        }   64    }   65    public static class UnitTest {   66        @Test   67        public void testPrimary() {   68            HystrixRequestContext context = HystrixRequestContext.initializeContext();   69            try {   70                ConfigurationManager.getConfigInstance().setProperty("primarySecondary.usePrimary", true);   71                assertEquals("responseFromPrimary-20", new CommandFacadeWithPrimarySecondary(20).execute());   72            } finally {   73                context.shutdown();   74                ConfigurationManager.getConfigInstance().clear();   75            }   76        }   77        @Test   78        public void testSecondary() {   79            HystrixRequestContext context = HystrixRequestContext.initializeContext();   80            try {   81                ConfigurationManager.getConfigInstance().setProperty("primarySecondary.usePrimary", false);   82                assertEquals("responseFromSecondary-20", new CommandFacadeWithPrimarySecondary(20).execute());   83            } finally {   84                context.shutdown();   85                ConfigurationManager.getConfigInstance().clear();   86            }   87        }   88    }   89}   复制代码

NOTE:显示调用降级适用于特殊需求的场景,fallback用于业务处理，fallback不再承担降级职责，建议慎重使用，会造成监控统计换乱等问题.

命令调用合并:HystrixCollapser

命令调用合并允许多个请求合并到一个线程/信号下批量执行。

执行流程图如下:

1public class CommandCollapserGetValueForKey extends HystrixCollapser
     
      
       , String, Integer> {    2    private final Integer key;    3    public CommandCollapserGetValueForKey(Integer key) {    4        this.key = key;    5    }    6    @Override    7    public Integer getRequestArgument() {    8        return key;    9    }   10    @Override   11    protected HystrixCommand
       
        
         > createCommand(final Collection
         
          
           > requests) {   12        //创建返回command对象   13        return new BatchCommand(requests);   14    }   15    @Override   16    protected void mapResponseToRequests(List
           
             batchResponse, Collection
            
             
              > requests) {   17        int count = 0;   18        for (CollapsedRequest
              
                request : requests) {   19            //手动匹配请求和响应   20            request.setResponse(batchResponse.get(count++));   21        }   22    }   23    private static final class BatchCommand extends HystrixCommand
               
                
                 > {   24        private final Collection
                 
                  
                   > requests;   25        private BatchCommand(Collection
                   
                    
                     > requests) {   26                super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))   27                    .andCommandKey(HystrixCommandKey.Factory.asKey("GetValueForKey")));   28            this.requests = requests;   29        }   30        @Override   31        protected List
                     
                       run() {   32            ArrayList
                      
                        response = new ArrayList
                       
                        ();   33            for (CollapsedRequest
                        
                          request : requests) {   34                response.add("ValueForKey: " + request.getArgument());   35            }   36            return response;   37        }   38    }   39    public static class UnitTest {   40        HystrixRequestContext context = HystrixRequestContext.initializeContext();   41        try {   42            Future
                         
                           f1 = new CommandCollapserGetValueForKey(1).queue();   43            Future
                          
                            f2 = new CommandCollapserGetValueForKey(2).queue();   44            Future
                           
                             f3 = new CommandCollapserGetValueForKey(3).queue();   45            Future
                            
                              f4 = new CommandCollapserGetValueForKey(4).queue();   46            assertEquals("ValueForKey: 1", f1.get());   47            assertEquals("ValueForKey: 2", f2.get());   48            assertEquals("ValueForKey: 3", f3.get());   49            assertEquals("ValueForKey: 4", f4.get());   50            assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());   51            HystrixCommand
                              command = HystrixRequestLog.getCurrentRequest().getExecutedCommands().toArray(new HystrixCommand
                             [1])[0];   52            assertEquals("GetValueForKey", command.getCommandKey().name());   53            assertTrue(command.getExecutionEvents().contains(HystrixEventType.COLLAPSED));   54            assertTrue(command.getExecutionEvents().contains(HystrixEventType.SUCCESS));   55        } finally {   56         context.shutdown();   57        }      58    }   59}   复制代码

NOTE:使用场景:HystrixCollapser用于对多个相同业务的请求合并到一个线程甚至可以合并到一个连接中执行，降低线程交互次和IO数,但必须保证他们属于同一依赖.

原文出自：

http://hot66hot.iteye.com/blog/2155036

觉得本文对你有帮助？请分享给更多人 关注「编程无界」，提升装逼技能