Hystrix熔断和限流(一)
Hystrix是熔断和限流最常用的第三方组件, 今天开始就一起从基本hystrix API入手看下如何做到熔断和限流的. Hystrix是使用命令模式, 将请求封装成命令, 并定义一系列针对命令的方法或操作.
01
HelloWorld
程序员惯例, 先看个hello world
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() {
return "Hello " + name + "!";
}
}示例中CommandHelloWorld是继承自HystrixCommand类, 处理逻辑为run(), 也是重写父类方法, 处理参数是以类构造参数的方式传入的实例变量.
每次都需要新实例化一个命令对象进行业务处理
new CommandHelloWorld("World");02
同步执行与异步执行
hystrix提供了同步和异步两种执行方法:
同步执行:
String result = new CommandHelloWorld("World").execute();异步执行:
Future<String> future = new CommandHelloWorld("World").queue();
String result = future.get();看下execute()方法可以发现, 同步执行也只是封装了异步等待的过程.
public R execute() {
try {
return queue().get();
} catch (Exception e) {
throw decomposeException(e);
}
}再看下queue()的实现方法, 可以发现hystrix的实现逻辑核心是toObservable()方法;
public Future<R> queue() {
final Observable<R> o = toObservable();
final Future<R> f = o.toBlocking().toFuture();
if (f.isDone()) {
try {
f.get();
return f;
} catch (Exception e) {
// ...
}
}
return f;
}03
观察者
Hystrix除了execute()和queue()两种执行执行方式,还提供了Observable的方式处理结果.
Observable<String> fWorld = new CommandHelloWorld("World").observe();
String result = fWorld.toBlocking().single();通过Observable对象, 还可以进行业务扩展 onNext(): 是run()逻辑执行完毕后的自定义逻辑, 参数为run()返回值. onCompleted(): run()处理完的回调方法 onError():异常处理回调
fWorld.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
// do something
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onNext(String v) {
// do something
}
});还有一种忽略异常和onCompleted()的处理方式
fWorld.subscribe(new Action1<String>() {
@Override
public void call(String v) {
// do something
}
});一起看下observe()方法, 可以发现逻辑核心也是toObservable()方法; toObservable()方法, 我们会后续详细说明是如何实现的.
public Observable<R> observe() {
ReplaySubject<R> subject = ReplaySubject.create();
toObservable().subscribe(subject);
return subject;
}04
Fallback
fallback是hystrix被广泛接受和使用的最主要原因之一, 下面一起看下fallback是如何使用的. Command逻辑执行失败时, 会调用getFallback()方法, 执行熔断逻辑.
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 + "!";
}
}05
缓存模式
Hystrix还提供了缓存模式, 在一次请求上下文中, 同样的数据多次请求时, 可以使用缓存处理方提高效率的. 缓存key可以根据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);
}
}开启hystrix请求上下文(HystrixRequestContext), 确定缓存使用范围
public void testWithCacheHits() {
HystrixRequestContext context = HystrixRequestContext.initializeContext();
try {
CommandUsingRequestCache command2a = new CommandUsingRequestCache(2);
CommandUsingRequestCache command2b = new CommandUsingRequestCache(2);
assertTrue(command2a.execute());
assertFalse(command2a.isResponseFromCache());
assertTrue(command2b.execute());
assertTrue(command2b.isResponseFromCache());
} finally {
context.shutdown();
}
}今天对hystrix的应用有了初步了解, 下次会详细分析下toObservable()方法, 了解hystrix是如何做到限流和fallback的. ?