Hystrix使用介绍

一时失言乱红尘 2021-06-24 15:56 579阅读 0赞

**1、执行方式**    
HystrixCommand提供了3种执行方式：

**1）同步执行**：即一旦开始执行该命令，当前线程就得阻塞着直到该命令返回结果，然后才能继续执行下面的逻辑。当调用命令的execute()方法即为同步执行， 示例：

@Test  
        public void synchronousExecute() throws Exception {  
            ThreadEchoCommand command = new ThreadEchoCommand("xianlinbox");  
            String result = command.execute();  
            assertThat(result,equalTo("Echo: xianlinbox"));  
        }

**2）异步执行**： 命令开始执行会返回一个Future<T>的对象，不阻塞后面的逻辑，开发者自己根据需要去获取结果。当调用HystrixCommand的queue()方法即为异步执行

@Test  
        public void asynchronousExecute() throws Exception {  
            ThreadEchoCommand command = new ThreadEchoCommand("xianlinbox");  
            Future<String> result = command.queue();  
            while (!result.isDone()){  
                System.out.println("Do other things ...");  
            }  
            assertThat(result.get(),equalTo("Echo: xianlinbox"));  
        }

**3）响应式执行**： 命令开始执行会返回一个Observable<T> 对象，开发者可以给给Obeservable对象注册上Observer或者Action1对象，响应式地处理命令执行过程中的不同阶段。当调用HystrixCommand的observe()方法，或使用Observable的工厂方法（just(),from()）即为响应式执行，这个功能的实现是基于Netflix的另一个开源项目RxJava（ [https://github.com/Netflix/RxJava][https_github.com_Netflix_RxJava] ）来的，更细节的用法可以参考： [https://github.com/Netflix/Hystrix/wiki/How-To-Use\#wiki-Reactive-Execution][https_github.com_Netflix_Hystrix_wiki_How-To-Use_wiki-Reactive-Execution] 。 示例：

@Test  
        public void reactiveExecute1() throws Exception {  
            ThreadEchoCommand command1 = new ThreadEchoCommand("xianlinbox");  
            Observable<String> result = command1.observe();  
            result.subscribe(new Action1<String>() {  
                @Override  
                public void call(String s) {  
                    logger.info("Command called. Result is:{}", s);  
                }1  
            });  
            Thread.sleep(1000);  
        }  
      
        @Test  
        public void reactiveExecute2() throws Exception {  
            ThreadEchoCommand command = new ThreadEchoCommand("xianlinbox");  
            Observable<String> result = command.observe();  
            result.subscribe(new Observer<String>() {  
                @Override  
                public void onCompleted() {  
                    logger.info("Command Completed");  
                }  
      
                @Override  
                public void onError(Throwable e) {  
                    logger.error("Command failled", e);  
                }  
      
                @Override  
                public void onNext(String args) {  
                    logger.info("Command finished,result is {}", args);  
                }  
            });  
            Thread.sleep(1000);  
        }

**2、隔离方式（Thread Pool和Semaphores）**    
Hystrix支持2种隔离方式：

**1）ThreadPool**：即根据配置把不同的命令分配到不同的线程池中，这是比较常用的隔离策略，该策略的优点是隔离性好，并且可以配置断路，某个依赖被设置断路之后，系统不会再尝试新起线程运行它，而是直接提示失败，或返回fallback值；缺点是新起线程执行命令，在执行的时候必然涉及到上下文的切换，这会造成一定的性能消耗，但是Netflix做过实验，这种消耗对比其带来的价值是完全可以接受的，具体的数据参见Hystrix Wiki（https://github.com/Netflix/Hystrix/wiki/How-it-Works\#wiki-Isolation）。 本文前面的例子都是使用的TheadPool隔离策略。   
  
**2、Semaphores**：信号量，顾名思义就是使用一个信号量来做隔离，开发者可以限制系统对某一个依赖的最高并发数。这个基本上就是一个限流的策略。每次调用依赖时都会检查一下是否到达信号量的限制值，如达到，则拒绝。该隔离策略的优点不新起线程执行命令，减少上下文切换，缺点是无法配置断路，每次都一定会去尝试获取信号量。示例：

public class SemaphoreEchoCommand extends HystrixCommand<String> {  
        private Logger logger = LoggerFactory.getLogger(ThreadEchoCommand.class);  
        private String input;  
      
        protected SemaphoreEchoCommand(String input) {  
            super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("Semaphore Echo"))  
                    .andCommandKey(HystrixCommandKey.Factory.asKey("Echo"))  
                    .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()  
                            .withExecutionIsolationStrategy(HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE)  
                            .withExecutionIsolationSemaphoreMaxConcurrentRequests(2)));  
            this.input = input;  
        }  
      
        @Override  
        protected String run() throws Exception {  
            logger.info("Run command with input: {}", input);  
            Thread.currentThread().sleep(100);  
            return "Echo: " + input;  
        }  
    }  
        @Test  
        public void semaphoresCommandExecute() throws Exception {  
            SemaphoreEchoCommand command = new SemaphoreEchoCommand("xianlinbox");  
            assertThat(command.execute(), equalTo("Echo: xianlinbox"));  
        }  
      
        @Test  
        public void semaphoresCommandMultiExecute() throws Exception {  
            for (int i = 0; i < 5; i++) {  
                final SemaphoreEchoCommand command = new SemaphoreEchoCommand("xianlinbox-" + i);  
                Thread thread = new Thread(new Runnable() {  
                    @Override  
                    public void run() {  
                        command.queue();  
                    }  
                });  
                thread.start();  
            }  
            Thread.sleep(1000);  
        }

第一个测试的运行日志如下：   
23:10:34.996 \[main\] INFO  c.n.c.DynamicPropertyFactory - DynamicPropertyFactory is initialized with configuration sources: com.netflix.config.ConcurrentCompositeConfiguration@2224df87   
23:10:35.045 \[main\] INFO  d.ThreadEchoCommand - Run command with input: xianlinbox   
从运行日志可以看到，HystrixCommand一样是在主线程中执行。   
  
第二个测试的运行日志如下：   
14:56:22.285 \[Thread-5\] INFO  d.ThreadEchoCommand - Run command with input: xianlinbox-4   
14:56:22.285 \[Thread-1\] INFO  d.ThreadEchoCommand - Run command with input: xianlinbox-0   
Exception in thread "Thread-2" Exception in thread "Thread-4" com.netflix.hystrix.exception.HystrixRuntimeException: Echo could not acquire a semaphore for execution and no fallback available.   
示例中，设置的信号量最大值为2， 因此可以看到有2个线程可以成功运行命令，第三个则会得到一个无法获取信号量的HystrixRuntimeException。

**3、优雅降级**   
在调用第三方服务时，总是无可避免会出现一些错误(fail,timeout等)，再加上上面提到的线程池大小，信号量的限制等等，在执行HystrixComamnd的过程中，总难免会抛出一些异常。而Hystrix为执行过程中的异常情况提供了优雅的降级方案,只需要在自己的HystrixCommand中实现getFallback()方法，当异常出现时，就会自动调用getFallback()方法的值. 示例：为第一小节中的AddressHystrixCommand和ContactHystrixCommand添加getFallback()方法， 当有异常发生的时候，直接返回null:

@Override  
        protected Contact getFallback() {  
            logger.info("Met error, using fallback value: {}", customerId);  
            return null;  
        }

然后，停掉Stub的Contact和Address服务， 再次调用GetCustomer服务（http://localhost:8080/HystrixDemo/customers/1），得到结果如下：   
\{"id":"1","name":"xianlinbox","contact":null,"address":null\}   
运行日志：   
15:22:08.847 \[hystrix-Contact-1\] INFO  c.x.h.d.ContactHystrixCommand - Get contact for customer 1   
15:22:09.098 \[hystrix-Contact-1\] INFO  c.x.h.d.ContactHystrixCommand - Met error, using fallback value: 1   
15:22:09.101 \[hystrix-Address-1\] INFO  c.x.h.d.AddressHystrixCommand - Get address for customer 1   
15:22:09.103 \[hystrix-Address-1\] INFO  c.x.h.d.AddressHystrixCommand - Met error, using fallback value: 1

**4、请求作用域特性**   
作用域设置：要想要使用请求作用域特性,首先必须把HystrixCommand置于HystrixRequestContext的生命周期管理中，其典型用法是在Web应用中增加一个ServletFilter，把每个用户Request用HystrixRequestContext包起来。示例：

public class HystrixRequestContextServletFilter implements Filter {  
        ...  
        public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)   
         throws IOException, ServletException {  
            //启动HystrixRequestContext     
            HystrixRequestContext context = HystrixRequestContext.initializeContext();  
            try {  
                chain.doFilter(request, response);  
            } finally {  
                //关闭HystrixRequestContext  
                context.shutdown();  
            }  
        }  
    }

然后把该SevletFilter配置到web.xml中：

<filter>  
            <display-name>HystrixRequestContextServletFilter</display-name>  
            <filter-name>HystrixRequestContextServletFilter</filter-name>  
            <filter-class>com.xianlinbox.hystrix.filter.HystrixRequestContextServletFilter</filter-class>  
        </filter>  
      
        <filter-mapping>  
            <filter-name>HystrixRequestContextServletFilter</filter-name>  
            <url-pattern>/*</url-pattern>  
        </filter-mapping>

设置了请求作用域之后，接下来看看，我们从中可以得到哪些好处：   
  
**1）请求缓存（Request Cache）** ：即当用户调用HystrixCommand时，HystrixCommand直接从缓存中取而不需要调用外部服务。HystrixCommand从缓存中取需要3个条件：   
1. 该HystrixCommand被包裹一个HystrixRequestContext中   
2. 该HystrixCommand实现了getCacheKey()方法   
3. 在HystrixRequestContext中已有相同Cache Key值的缓存

public void requestCache() throws Exception {  
            HystrixRequestContext context = HystrixRequestContext.initializeContext();  
            try {  
                ThreadEchoCommand command1 = new ThreadEchoCommand("xianlinbox");  
                ThreadEchoCommand command2 = new ThreadEchoCommand("xianlinbox");  
      
                assertThat(command1.execute(),equalTo("Echo: xianlinbox"));  
                assertThat(command1.isResponseFromCache(),equalTo(false));  
                assertThat(command2.execute(),equalTo("Echo: xianlinbox"));  
                assertThat(command2.isResponseFromCache(),equalTo(true));  
            } finally {  
                context.shutdown();  
            }  
      
            context = HystrixRequestContext.initializeContext();  
            try {  
                ThreadEchoCommand command3 = new ThreadEchoCommand("xianlinbox");  
                assertThat(command3.execute(),equalTo("Echo: xianlinbox"));  
                assertThat(command3.isResponseFromCache(),equalTo(false));  
            } finally {  
                context.shutdown();  
            }  
        }

从上面的例子看以得到，一旦重新初始化了RequestContext，Cache也都全部失效了。另外，从Cache中获取值不会去执行HystrixCommand的run()方法。   
  
除了重新初始化RequestContext，Hystrix还提供了另外一种方式来刷新Cache，该方式需要使用HystrixRequestCache的clear()方法，示例：在ThreadEchoCommand中实现一个静态方法flushCache( )，该方法会调用HystrixRequestCache的clear方法清理Cache

public static void flushCache(String cacheKey) {  
           HystrixRequestCache.getInstance(HystrixCommandKey.Factory.asKey("Echo"),  
                   HystrixConcurrencyStrategyDefault.getInstance()).clear(cacheKey);  
       }  
         
       @Test  
       public void flushCacheTest() throws Exception {  
           HystrixRequestContext context = HystrixRequestContext.initializeContext();  
           try {  
               ThreadEchoCommand command1 = new ThreadEchoCommand("xianlinbox");  
               ThreadEchoCommand command2 = new ThreadEchoCommand("xianlinbox");  
      
               assertThat(command1.execute(), equalTo("Echo: xianlinbox"));  
               assertThat(command1.isResponseFromCache(), equalTo(false));  
               assertThat(command2.execute(), equalTo("Echo: xianlinbox"));  
               assertThat(command2.isResponseFromCache(), equalTo(true));  
      
               ThreadEchoCommand.flushCache("xianlinbox");  
               ThreadEchoCommand command3 = new ThreadEchoCommand("xianlinbox");  
               assertThat(command3.execute(), equalTo("Echo: xianlinbox"));  
               assertThat(command3.isResponseFromCache(), equalTo(false));  
           } finally {  
               context.shutdown();  
           }  
       }

通过这个机制,开发者可以实现Get-Set-Get的Cache验证机制，防止因为Cache导致的不一致状况。   
  
**2）批量执行请求(Request Collapsing)** ：即用户可以把多个命令封装到一个HystrixCommand中执行以提升效率，这多个命令会在一个线程中依次执行（注：经笔者测试，在JDK6下线程数固定，但是在JDK7下的运行线程数不固定）。要使用该特性需要把依赖调用封装到一个HystrixCollapser<BatchReturnType,ResponseType,RequestArgumentType>中， 该抽象类的主要作用有3个：   
1. 把所有的依赖调用封装到一个CollapseRequest的集合中   
2. 以第一步得到的CollapseRequest集合为参数创建一个HystrixCommand   
3. 把第二步得到的结果集一一对应的设置到对应的CollapseRequest中   
  
为了支持上面的功能,该抽象类提供了3个泛型参数：   
BatchReturnType：即BatchCommand的返回值，通常为ResponseType的集合。   
ResponseType：依赖调用的返回值。   
RequestArgumentType：依赖调用的参数，如果有多个参数，需封装为一个对象或使用集合。   
示例：

public class CollapseEchoHystrixCommand extends HystrixCollapser<List<String>, String, String> {  
        private Logger logger = LoggerFactory.getLogger(CollapseEchoHystrixCommand.class);  
        private String input;  
      
        public CollapseEchoHystrixCommand(String input) {  
            super(HystrixCollapser.Setter  
                    .withCollapserKey(HystrixCollapserKey.Factory.asKey("Echo Collapse")));  
            this.input = input;  
        }  
      
        @Override  
        public String getRequestArgument() {  
            return input;  
        }  
      
        @Override  
        protected HystrixCommand<List<String>> createCommand(Collection<CollapsedRequest<String, String>> collapsedRequests) {  
            return new BatchCommand(collapsedRequests);  
        }  
      
        @Override  
        protected void mapResponseToRequests(List<String> batchResponse, Collection<CollapsedRequest<String, String>> collapsedRequests) {  
            logger.info("Mapping response to Request");  
            int count = 0;  
            for (CollapsedRequest<String, String> request : collapsedRequests) {  
                request.setResponse(batchResponse.get(count++));  
            }  
      
        }  
      
        private class BatchCommand extends HystrixCommand<List<String>> {  
            private Collection<CollapsedRequest<String, String>> requests;  
      
            public BatchCommand(Collection<CollapsedRequest<String, String>> requests) {  
                super(HystrixCommandGroupKey.Factory.asKey("Batch"));  
                this.requests = requests;  
            }  
      
            @Override  
            protected List<String> run() throws Exception {  
                logger.info("Run batch command");  
                List<String> responses = new ArrayList<String>();  
                for (CollapsedRequest<String, String> request : requests) {  
                    logger.info("Run request: {}", request.getArgument());  
                    responses.add("Echo: " + request.getArgument());  
                }  
                return responses;  
            }  
        }  
    }  
      
        @Test  
        public void collapseCommandTest() throws Exception {  
            HystrixRequestContext context = HystrixRequestContext.initializeContext();  
      
            try {  
                Future<String> result1 = new CollapseEchoHystrixCommand("xianlinbox-1").queue();  
                Future<String> result2 = new CollapseEchoHystrixCommand("xianlinbox-2").queue();  
                Future<String> result3 = new CollapseEchoHystrixCommand("xianlinbox-3").queue();  
      
                assertThat(result1.get(),equalTo("Echo: xianlinbox-1"));  
                assertThat(result2.get(),equalTo("Echo: xianlinbox-2"));  
                assertThat(result3.get(),equalTo("Echo: xianlinbox-3"));  
      
                assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());  
            } finally {  
                context.shutdown();  
            }  
        }

运行日志：   
03:10:58.584 \[main\] INFO  d.CollapseEchoHystrixCommand - Get argument   
03:10:58.597 \[main\] INFO  d.CollapseEchoHystrixCommand - Get argument   
03:10:58.597 \[main\] INFO  d.CollapseEchoHystrixCommand - Get argument   
03:10:58.598 \[HystrixTimer-1\] INFO  d.CollapseEchoHystrixCommand - Create batch command   
03:10:58.637 \[hystrix-Batch-1\] INFO  d.CollapseEchoHystrixCommand - Run batch command   
03:10:58.637 \[hystrix-Batch-1\] INFO  d.CollapseEchoHystrixCommand - Run request: xianlinbox-1   
03:10:58.639 \[hystrix-Batch-1\] INFO  d.CollapseEchoHystrixCommand - Run request: xianlinbox-2   
03:10:58.639 \[hystrix-Batch-1\] INFO  d.CollapseEchoHystrixCommand - Run request: xianlinbox-3   
03:10:58.644 \[RxComputationThreadPool-1\] INFO  d.CollapseEchoHystrixCommand - Mapping response to Request   
  
从运行日志可以看到，整个Collapser的运行过程：   
1. 获取调用参数，封装到CollapseRequest中   
2. 以封装后的List<CollapseRequest>为参数创建Batch HystrixComand   
3. Batch HystrixCommand运行所有的请求，把所有的返回放到List<Response>中   
4. 把Response设置到对应的CollapseRequest中，返回给调用者。

[https_github.com_Netflix_RxJava]: https://github.com/Netflix/RxJava
[https_github.com_Netflix_Hystrix_wiki_How-To-Use_wiki-Reactive-Execution]: https://github.com/Netflix/Hystrix/wiki/How-To-Use#wiki-Reactive-Execution