Netflix Hystrix笔记
maven引入
<dependency>
<groupId>com.netflix.hystrix</groupId>
<artifactId>hystrix-core</artifactId>
<version>1.5.13</version>
</dependency>hello,world
- 继承HystrixCommand；
- HystrixCommandGroupKey是基类构造器必须的一个参数，可以实现接口来创建，也可以通过工厂方法HystrixCommandGroupKey.Factory.asKey(String)来创建；
- 实现run()方法；
- 调用
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() {
// a real example would do work like a network call here
return "Hello " + name + "!";
}
}
@Test
public void testSynchronous() {
assertEquals("Hello World!", new CommandHelloWorld("World").execute();
}execute/queue/observe
String s1 = new CommandHelloWorld("World").execute();
Future<String> fs = new CommandHelloWorld("World").queue();
String s2 = fs.get();
Observable<String> ho = new CommandHelloWorld("World").observe();
ho.subscribe(new Action1<String>() {
@Override
public void call(String s) {
// value emitted here
}
});
Observable<String> co = new CommandHelloWorld("World").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
Fallback
public class CommandHelloFailure extends HystrixCommand<String> {
private final String name;
public CommandHelloFailure(String name) {
super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"));
this.name = name;
}
@Override
protected String run() {
throw new RuntimeException("this command always fails");
}
@Override
protected String getFallback() {
return "Hello Failure " + name + "!";
}
}Command Name & Command Group & Command Thread-Pool
public CommandHelloWorld(String name) {
super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"))
.andCommandKey(HystrixCommandKey.Factory.asKey("HelloWorld"))
.andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("HelloWorldPool")));
this.name = name;
}概念说明
- Command Group: Hystrix uses the command group key to group together commands such as for reporting, alerting, dashboards, or team/library ownership.
- Command Thread-Pool: The thread-pool key represents a HystrixThreadPool for monitoring, metrics publishing, caching, and other such uses. A HystrixCommand is associated with a single HystrixThreadPool as retrieved by the HystrixThreadPoolKey injected into it, or it defaults to one created using the HystrixCommandGroupKey it is created with.
缓存
通过实现getCacheKey()方法来开启缓存
public class CommandUsingRequestCache extends HystrixCommand<Boolean> {
private final int value;
protected CommandUsingRequestCache(int value) {
super(HystrixCommandGroupKey.Factory.asKey("ExampleGroup"));
this.value = value;
}
@Override
protected Boolean run() {
return value == 0 || value % 2 == 0;
}
@Override
protected String getCacheKey() {
return String.valueOf(value);
}
}请求合并
request-scoped: 同一个HystrixRequestContext
globally-scoped: 跨多个HystrixRequestContext
关于请求合并这篇通过HystrixCollapser合并请求提高应用吞吐量的例子比较好，正好利用请求合并并发挥es的mget的作用。转载于此。
//利用hystrix合并请求
// scope:
// 合并作用域，默认是REQUEST，就是不会跨越多个请求会话的，只在当前用户请求中合并多次请求为批处理请求。这里改成GLOBAL，就是可以跨越request context，合并不同用户的请求为一次批处理请求。
// timerDelayInMilliseconds & maxRequestsInBatch:
// 表示合并请求的时间窗口为50ms,窗口时间内最多合并200个请求。默认值是10ms，合并请求数不限。
@HystrixCollapser(batchMethod = "mGetDataFromEs", scope = GLOBAL, collapserProperties = {
@HystrixProperty(name = "timerDelayInMilliseconds", value = "50"),
@HystrixProperty(name = "maxRequestsInBatch", value = "200"),
})
@Cacheable(value = "esRequestCache", cacheManager = "esRequestCacheManager", key = "#doc.index+':'+#doc.type+':'+#doc.id", unless = "#result == null || #result.size() ==0")
public List<RecommendItem> getDataFromEs3(EsDocGetRequest doc) {
return null;
}
// fallbackMethod: 指定fallback方法
@HystrixCommand(fallbackMethod = "mGetDataFromEsFallback")
public List<List<RecommendItem>> mGetDataFromEs(List<EsDocGetRequest> docs) throws IOException {
logger.debug("================mGetDataFromEs==============");
List<List<RecommendItem>> datas = new ArrayList<>();
List<Doc> docList = new ArrayList<>();
for (EsDocGetRequest req : docs) {
docList.add(new Doc(req.getIndex(), req.getType(), req.getId()));
}
long start = System.currentTimeMillis();
JestResult result = jestClient.execute(new MultiGetExt.Builder.ByDoc(docList).build());
logger.info("============mGetDataFromEs cost time:{}", (System.currentTimeMillis() - start));
// assertTrue(result.getErrorMessage(), result.isSucceeded());
if (result.isSucceeded()) {
JsonArray actualDocs = result.getJsonObject().getAsJsonArray("docs");
for (int i = 0; i < docs.size(); i++) {
boolean hasError = actualDocs.get(i).getAsJsonObject().has("error");
boolean found = !hasError && actualDocs.get(i).getAsJsonObject().getAsJsonPrimitive("found").getAsBoolean();
if (found) {
EsRcmdResult esRcmdResult = gson.fromJson(actualDocs.get(i).getAsJsonObject().get("_source"), EsRcmdResult.class);
datas.add(esRcmdResult.getData());
} else {
datas.add(Collections.emptyList());
}
}
return datas;
} else {
throw new ServiceException(result.getErrorMessage());
}
}
public List<List<RecommendItem>> mGetDataFromEsFallback(List<EsDocGetRequest> docs) throws IOException {
logger.error("============mGetDataFromEs fallback=============");
return Collections.emptyList();
}Request Context Setup
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();
}
}
}Common Patterns
In the following sections are common uses and patterns of use for HystrixCommand and HystrixObservableCommand.
Fail Fast
- has no fallback behivor
- throws HystrixRuntimeException
- 通过HystrixRuntimeException.getCause获取原始异常对象
Fail Silent
Failing silently is the equivalent of returning an empty response or removing functionality. It can be done by returning null, an empty Map, empty List, or other such responses.
@Override
protected String getFallback() {
return null;
}
@Override
protected List<String> getFallback() {
return Collections.emptyList();
}
// HystrixObservableCommand
@Override
protected Observable<String> resumeWithFallback() {
return Observable.empty();
}Fallback: Static
在fallback方法里返回一个静态值
@Override
protected Boolean getFallback() {
return true;
}
@Override
protected Observable<Boolean> resumeWithFallback() {
return Observable.just( true );
}Falllback: Stubbed
You typically use a stubbed fallback when your command returns a compound object containing multiple fields, some of which can be determined from other request state while other fields are set to default values.
Fallback: Cache via Network
Dynamic Property
example:primary--secondary-with-fallback
// write
ConfigurationManager.getConfigInstance().setProperty("primarySecondary.usePrimary", true);
// read
DynamicBooleanProperty usePrimary = DynamicPropertyFactory.getInstance().getBooleanProperty("primarySecondary.usePrimary", true);Properties Strategy
If you implement a custom HystrixPropertiesStrategy, this gives you full control over how properties are defined for the system.
The default implementation uses Archaius.
todo: 如何运行时变更配置？HystrixPlugins.getInstance().registerPropertiesStrategy(null);?
hystrix-javanica
什么是ReactiveX？
ReactiveX is a library for composing asynchronous and event-based programs by using observable sequences.
It extends the observer pattern to support sequences of data and/or events and adds operators that allow you to compose sequences together declaratively while abstracting away concerns about things like low-level threading, synchronization, thread-safety, concurrent data structures, and non-blocking I/O.
其他
- javadoc: HystrixRequestContext
- hystrix-request-servlet