标签：ext group 通过 start 开头 netflix cti 映射 stream
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如何做到这点?
下面用几个简单的例子进行说明。
通过几个简单的例子，对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中调用。
说明下上面的例子中出现的两个类。
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()而已。
尝试执行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 onError(Throwable e) {
System.out.printf("error...%s\n", e.getMessage());
}
@Override
public void onNext(String s) {
System.out.printf("from next...%s\n", s);
}
});
}
}
试试在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。
接下来说说Command是如何和断路器(circuit breaker)交互的。
HystrixCommand属性及默认值可以参考抽象类HystrixCommandProperties，其中以circuitBreaker开头为断路器相关属性。
这里先列出3个关于断路器的属性，分别为:
工作流程大致如下:
之前并没有注意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。
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个泛型类型，依次为
继承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。
事实上，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();
}
}
}标签：ext group 通过 start 开头 netflix cti 映射 stream
原文地址：http://www.cnblogs.com/kavlez/p/hytrix-getting-started.html
