首页 > 代码库 > Netflix Hystrix - 快速入门

Netflix Hystrix - 快速入门

2024-09-21 21:38:26   211人阅读

Hystrix最初是由Netflix的API team研发的，用于提高API的弹性和性能，2012年在公司内部广受好评。


如果你的应用是一个单独的应用，那几乎不用在意断路的问题。

但在分布式环境中，各个应用错综复杂的依赖关系，一个不稳定的服务会拖累依赖它的服务。


简单来说，就是将服务之间的访问隔离开来，在错误(包括超时)被传播之前拦截下来，并提供相应的处理逻辑，让这个分布式应用更有弹性。

Hystrix就是用来解决这一问题的lib，帮助开发者更方便的控制服务之间的通信。


在分布式系统中，你使用的第三方RPC API可能会提供服务通信拦截的功能，但通常不会涉及方方面面，更不能单独拿出来给其他API使用。

而Hystrix会提供这些:

  • 为服务通信提供保护、容错
  • 在复杂的依赖关系链中阻止错误传播
  • 快速失败
  • 根据具体事件进行毁掉
  • 支持降级
  • 近实时监控

 

假设你的分布式系统中存在几十甚至上百个服务，即使每个服务都能保证99.99的可用性，但是在依赖关系错综复杂，单个服务依赖数量过多，随着请求数量的上升，带来的损失也是惨重的。


如果我为这个服务本身设置了超时时间，该服务对于不同的依赖方的权重不尽相同。

假设服务A和B都依赖服务C。对于A，它可能依赖很多服务，但C无法在1秒内响应时就放弃。而对于B，C是至关重要的服务，除非是业务数据异常，否则绝对不能中途停止。


如果C设置的超时时间为30s，那么A和B则同样需要等待30s，这显然是不合理的。而等待中的这些请求会耗费什么资源就看具体情况了，最坏的情况是拖垮了整个应用。

因此，延迟(lagency)和失败(failure)都需要被隔离。


Hystrix如何做到这点?

  • 通过HystrixCommand在独立的线程调用服务。
  • 超时时间由调用方掌握。
  • 为每个依赖维护一个小线程池，线程池满时可以拒绝请求，而不是将请求放入队列。
  • 区分事件，比如successe、failure, timeout、rejection，针对不同事件进行相应的回调。
  • 当失败占比超过指定阈值时启动断路器(circuit-breaker)，一段时间内阻止对特定依赖访问。

 

下面用几个简单的例子进行说明。

  

Getting Started

通过几个简单的例子，对Hystrix有个粗浅的认识。

首先，添加以下依赖

compile group: ‘com.netflix.hystrix‘, name: ‘hystrix-core‘, version: ‘1.5.10‘

 

参考如下main

package com.kavlez.lab.hystrix;import com.netflix.hystrix.HystrixCommand;import com.netflix.hystrix.HystrixCommandGroupKey;/** * @author Kavlez */public class Hello {    public static void main(String[] args) {        HystrixCommandGroupKey groupKey = HystrixCommandGroupKey.Factory.asKey("ExampleGroup");        HystrixCommand<String> hystrixCommand = new HystrixCommand<String>(groupKey) {            @Override            protected String run() throws Exception {                return "hi";            }        };        System.out.printf("exec command ... result = %s", hystrixCommand.execute());    }}

 

被覆写的run()为HystrixCommand中定义的抽象方法，调用依赖服务时也是在run中调用。

说明下上面的例子中出现的两个类。

  • HystrixCommand: 用command的包含任何潜在风险(延时、失败)的代码，进而对其进行处理，比如容错、统计、断路...

  • HystrixCommandGroupKey: 所谓command的group，用于对一系列command统一进行一些操作。

    A group name for a {@link HystrixCommand}. This is used for grouping together commands such as for reporting, alerting, dashboards or team/library ownership.

 

和group一样，command也是有名称的。默认为类名

getClass().getSimpleName();

 

但并没有提供相应的setter，只是提供了一个构造方法

protected HystrixCommand(Setter setter)

 

因此，如需指定command名称，参考如下

final HystrixCommand.Setter setter =        HystrixCommand.Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))                .andCommandKey(HystrixCommandKey.Factory.asKey("ExampleCommand"));HystrixCommand<String> hystrixCommand = new HystrixCommand<String>(setter) {    //...};

 

异步执行

如上面的例子中，我们可以通过execute()执行command，这是一种同步执行方式。

如果需要异步执行，只需要用queue()替代execute()即可。

Future<String> future = hystrixCommand.queue();try {    System.out.printf("exec command ... result = %s\n", future.get());} catch (ExecutionException e) {    e.printStackTrace();}

 

事实上，execute()不过是queue().get()而已。

 

Observe

尝试执行gradle dependencies，打印如下

compile - Dependencies for source set ‘main‘.+--- org.slf4j:slf4j-api:1.7.21\--- com.netflix.hystrix:hystrix-core:1.5.10     +--- org.slf4j:slf4j-api:1.7.0 -> 1.7.21     +--- com.netflix.archaius:archaius-core:0.4.1     |    +--- commons-configuration:commons-configuration:1.8     |    |    +--- commons-lang:commons-lang:2.6     |    |    \--- commons-logging:commons-logging:1.1.1     |    \--- org.slf4j:slf4j-api:1.6.4 -> 1.7.21     +--- io.reactivex:rxjava:1.2.0     \--- org.hdrhistogram:HdrHistogram:2.1.9

 

我想说的是hystrix依赖RxJava。

其中observe是比较典型的用法，HystrixCommand提供了两种方法observe和toObservable，官方对两者描述如下。

  • observe() — returns a “hot” Observable that executes the command immediately, though because the Observable is filtered through a ReplaySubject you are not in danger of losing any items that it emits before you have a chance to subscribe
  • toObservable() — returns a “cold” Observable that won’t execute the command and begin emitting its results until you subscribe to the Observable

 

显然，如果是通过toObservable，同一个command实例是无法被subscribe多次的。

尽管两者都返回Observable对象，但行为上稍有区别。


但本质上observe()几乎等同于toObservable().subscribe(subject) 

下面是一段例子，由于是通过observe()，命令可以有多个subscriber:

package com.kavlez.lab.hystrix;import com.netflix.hystrix.HystrixCommand;import com.netflix.hystrix.HystrixCommandGroupKey;import rx.Observable;import rx.Observer;/** * @author Kavlez */public class HelloObservable {    public static void main(String[] args) {        HystrixCommandGroupKey groupKey = HystrixCommandGroupKey.Factory.asKey("ExampleGroup");        HystrixCommand<String> hystrixCommand = new HystrixCommand<String>(groupKey) {            @Override            protected String run() throws Exception {                return "hi";            }        };        Observable<String> observe = hystrixCommand.observe();        observe.subscribe(s -> {            System.out.printf("from action1...%s\n", s);        });        observe.subscribe(new Observer<String>() {            @Override            public void onCompleted() {                System.out.println("completed...");            }            @Override            public void one rror(Throwable e) {                System.out.printf("error...%s\n", e.getMessage());            }            @Override            public void onNext(String s) {                System.out.printf("from next...%s\n", s);            }        });    }}

 

Fallback

试试在run中写一段Thread.sleep(1000)，或者加个断点让程序暂停一段时间。
出现j.u.c.TimeoutException和HystrixRuntimeException，且后者提示

timed-out and no fallback available.

 

这是因为HystrixCommand的getFallback()默认为

protected R getFallback() { throw new UnsupportedOperationException("No fallback available."); }

 

既然如此，我们只需要覆写该方法就可以实现降级(degradation)。

比如，把之前的例子改为:

HystrixCommand<String> hystrixCommand = new HystrixCommand<String>(groupKey) {    @Override    protected String run() throws Exception {        Thread.sleep(1000);        return "hi";    }    @Override    protected String getFallback() {        return "hi, sorry i am late...";    }};

 

getFallback并没有提供参数，这意味着fallback不止发生在timeout一种情况，failure、timeout、thread pool rejection都可以触发fallback。

 

Circuit Breaker

接下来说说Command是如何和断路器(circuit breaker)交互的。

HystrixCommand属性及默认值可以参考抽象类HystrixCommandProperties，其中以circuitBreaker开头为断路器相关属性。 

这里先列出3个关于断路器的属性，分别为:

  • HystrixCommandProperties.circuitBreakerRequestVolumeThreshold() : 请求容量阈值
  • HystrixCommandProperties.circuitBreakerErrorThresholdPercentage(): 错误占比阈值
  • HystrixCommandProperties.circuitBreakerSleepWindowInMilliseconds() : 状态时长

 

工作流程大致如下:

  1. 假设达到了request volume threshold，也就是metrics.healthCounts.totalCount大于该项
  2. 并且失败次数的占比也达到了error percentage，默认为50%
  3. 此时，断路器的状态从CLOSED变为OPEN
  4. 断路器状态变为OPEN后，接收到的请求将全部断路
  5. 过了恢复时间后，也就是sleep window in milliseconds(默认为5s)，断路器从OPEN变为HALF-OPEN状态。
  6. 如果变更为HALF-OPEN后的下一次请求失败，则变回OPEN状态，反之为CLOSED。

 

Request Cache

之前并没有注意Hystrix也提供了这样一个特性，command中可以通过覆写getCacheKey对请求进行缓存。

该方法默认返回null，也就是不缓存。

如果n个命令都在同一个request scope，则只有一个命令会被执行，其余n-1都是缓存。

 

代码参考如下

package com.kavlez.lab.hystrix;import com.netflix.hystrix.HystrixCommand;import com.netflix.hystrix.HystrixCommandGroupKey;import com.netflix.hystrix.HystrixCommandKey;import com.netflix.hystrix.strategy.concurrency.HystrixRequestContext;/** * @author Kavlez */public class ReqCache {    static class HelloCommand extends HystrixCommand<String> {        private static final HystrixCommand.Setter setter =                HystrixCommand.Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))                        .andCommandKey(HystrixCommandKey.Factory.asKey("ExampleCommand"));        private String requestKey;        protected HelloCommand(String requestKey) {            super(setter);            this.requestKey = requestKey;        }        @Override        protected String run() throws Exception {            return null;        }        @Override        protected String getCacheKey() {            return this.requestKey;        }    }    public static void main(String[] args) {        HystrixRequestContext.initializeContext();        HelloCommand hello1 = new HelloCommand("Billy");        hello1.execute();        System.out.println(hello1.isResponseFromCache());        hello1 = new HelloCommand("Billy");        hello1.execute();        System.out.println(hello1.isResponseFromCache());        hello1 = new HelloCommand("Van");        hello1.execute();        System.out.println(hello1.isResponseFromCache());    }}

 

注意这一行

HystrixRequestContext.initializeContext();

缺少HystrixRequestContext会提示illegal state。

 

Request Collapsing

Hystrix提供了一个叫collapse的特性，将多个请求进行合并，方便将多个请求限制在同一个time windows。

官方给出的例子是获取收藏夹中的300部电影，类似的场景确实常见。



或者再复杂一点，比如我要获取300部电影的工作人员信息，几部不同电影很可能存在相同的工作人员。

也许我可以...首先获取300部电影的列表，对其进行循环并get工作人员列表，然后再对其进行循环，依次请求工作人员的REST API...无论列表中是否有多个相同的电影和工作人员。

或者我可以仅仅为了这样的应用场景而专门设计一套API，专门用于获取电影列表中每一部电影的工作人员的信息。

但这显然是个笨方法，默许这样的方法会导致莫名其妙的API越来越多。

因此，为了应付这样的场景而抽象出一层collapsing layer是值得的。

这样一来，REST API和实体类可以依然保持单纯，而开发者只需要使用HystrixCollapser即可。

假设多个命令同时进行相同的请求，collapser可以将请求进行合并，批量请求，并将结果分发给各个命令。

 

参考如下例子

package com.kavlez.lab.hystrix;import com.google.common.collect.Lists;import com.google.common.collect.Maps;import com.netflix.hystrix.HystrixCollapser;import com.netflix.hystrix.HystrixCollapserKey;import com.netflix.hystrix.HystrixCommand;import com.netflix.hystrix.HystrixCommandGroupKey;import com.netflix.hystrix.strategy.concurrency.HystrixRequestContext;import java.util.Collection;import java.util.List;import java.util.Map;import java.util.concurrent.ExecutionException;import java.util.concurrent.Future;import java.util.stream.Collectors;/** * @author Kavlez * @since 5/12/17. */public class CollapserExample {    private static final String[] names = {"Protos", "Terran", "Hulk", "Anderson", "Uriah", "Gegard", "Velasquez", "Mcgregor", "Jose"};    static class User {        private int id;        private int code;        private String name;        public User(int id, int code, String name) {            this.id = id;            this.code = code;            this.name = name;        }        public int getId() {            return id;        }        public void setId(int id) {            this.id = id;        }        public int getCode() {            return code;        }        public void setCode(int code) {            this.code = code;        }        public String getName() {            return name;        }        public void setName(String name) {            this.name = name;        }    }    static class UserCollapser extends HystrixCollapser<Map<Integer, User>, User, Integer> {        private int userId;        private static final Setter setter = Setter.withCollapserKey(HystrixCollapserKey.Factory.asKey("UserCollapser"));        public UserCollapser(int userId) {            super(setter);            this.userId = userId;        }        @Override        public Integer getRequestArgument() {            return this.userId;        }        @Override        protected HystrixCommand<Map<Integer, User>> createCommand(                Collection<CollapsedRequest<User, Integer>> collapsedRequests) {            return new UserBatchCommand(collapsedRequests.stream().map(request -> {                System.out.println("arg mapped...");                return request.getArgument();            }).collect(Collectors.toList()));        }        @Override        protected void mapResponseToRequests(                Map<Integer, User> batchResponse, Collection<CollapsedRequest<User, Integer>> collapsedRequests) {            for (CollapsedRequest<User, Integer> request : collapsedRequests) {                Integer userId = request.getArgument();                request.setResponse(batchResponse.get(userId));            }        }    }    static class UserBatchCommand extends HystrixCommand<Map<Integer, User>> {        private List<Integer> ids;        private final static Setter setter = Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("UserBatchGroup"));        public UserBatchCommand(List<Integer> ids) {            super(setter);            this.ids = ids;        }        @Override        protected Map<Integer, User> run() throws Exception {            return this.getUsers();        }        Map<Integer, User> getUsers() {            Map<Integer, User> users = Maps.newHashMap();            for (Integer id : ids) {                int randomCode = (int) (Math.random() * 100);                users.put(id, new User(id, randomCode, names[randomCode % names.length]));            }            return users;        }    }    public static void main(String[] args) throws ExecutionException, InterruptedException {        HystrixRequestContext.initializeContext();        List<Future<User>> futures = Lists.newArrayList(1, 1, 1, 1, 2, 2, 2, 3, 3, 4).stream()                .map(userId -> new UserCollapser(userId).queue()).collect(Collectors.toList());        for (Future<User> future : futures) {            future.get();        }    }}

 

覆写HystrixCollapser时指定的3个泛型类型，依次为

  • batch command返回类型
  • response类型
  • request参数类型

 

继承HystrixCollapser需要覆写3个方法，分别为

  • protected abstract HystrixCommand createCommand(Collection<CollapsedRequest<ResponseType, RequestArgumentType>> requests); 工厂方法，用于创建HystrixCommand对象，或者说是一个专门用于处理批量请求的command。
    多数情况下，该方法创建的命令执行一次后就没什么用了，所以通常返回一个新的实例。
    由于是用于处理批量请求，所以通常会把CollapsedRequest集合整个传给command。

  • public abstract RequestArgumentType getRequestArgument();通过该方法来提供传递给HystrixCommand的参数，如果你需要传递多个参数，则封装到一个对象即可。

  • protected abstract void mapResponseToRequests(BatchReturnType batchResponse, Collection<CollapsedRequest<ResponseType, RequestArgumentType>> requests);createCommand创建了对应的command，该command结束后会调用mapResponseToRequests，该方法将BatchReturnType映射为Collection<CollapsedRequest<ResponseType, RequestArgumentType>> requests。

 

Request Context Setup

上面提到的内容都涉及到了request context。

事实上，Hystrix的一些功能都需要request context，也就是request-scoped features。

比如，上面的例子中的main方法中都有这么一行

HystrixRequestContext context = HystrixRequestContext.initializeContext();

 

这就需要开发者按需管理HystrixRequestContext的生命周期。

而request多是在web应用中比较常见，比如实现一个servlet filter，在doFilter方法中进行管理。

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();        }    }}

Netflix Hystrix - 快速入门


联系
我们