使用继承的方式来创建Hystrix 命令,不是用注解的方式来使用 Hystrix
-
创建 HelloGetCommand 对象,继承与 HystrixCommand 并实现 run 方法,示例如下:
public class HelloGetCommand extends HystrixCommand<String> {
private RestTemplate restTemplate;
public
HelloGetCommand(Setter setter, RestTemplate restTemplate) {super(setter);
this.restTemplate= restTemplate;
}
@Override
protected String run() throws Exception {
ResponseEntity<String> responseEntity=
restTemplate(getForEntity("http://ORG.DRSOFT.WEBSERVICE.HELLOSERVICE/hello/get", String.class);
return responseEntity.getBody();
}
@Override
protected String getFallback() {
return
"error id=" + id + "\tname=" + name;}
}
-
或者创建 HelloGetObserveableCommand 对象,继承于 HystrixObservableCommand 对象,并实现 construct 方法,如果希望是异步订阅,则在
new Observable.OnSubscribe<String>()
创建实例后,调用 subscribeOn 方法来设置,示例如下:public class HelloGetObservableCommand extends HystrixObservableCommand<String> {
private RestTemplate restTemplate;
public
HelloGetCommand(Setter setter, RestTemplate restTemplate) {super(setter);
this.restTemplate= restTemplate;
}
@Override
protected Observable<String> construct() {
return Observable.create(new Observable.OnSubscribe<String>() {
@Override
public
void
call(Subscriber<? super String> subscriber) {try {
if (subscriber.isUnsubscribed() == false) {
ResponseEntity<String> responseEntity=
restTemplate.getForEntity("http://ORG.DRSOFT.WEBSERVICE.HELLOSERVICE/hello/get",
String.class);
subscriber.onNext(responseEntity.getBody(););
subscriber.onCompleted();
}
} catch (Exception ex) {
subscriber.onError(ex);
}
}
});
}
@Override
protected Observable<String> resumeWithFallback() {
return Observable.create(new Observable.OnSubscribe<String>() {
@Override
public
void
call(Subscriber<? super String> subscriber) {subscriber.onNext("error id=" + id + "\tname=" + name);
subscriber.onCompleted();
}
});
}
}
-
HystrixCommand 对象实现了两种执行方法,包含了同步执行和异步执行,如下:
- exeucte():同步执行,从依赖的服务返回一个单一的结果对象,或是在发生错误的时候抛出异常,如果重写了 getFallback 方法,则会调用该方法,execute() 其实是 queue().get() 调用,只有 HystrixCommand 支持
- queue():异步执行,直接返回一个 Future 对象,其中包含了服务执行结束时要返回的单一结果对象,实际上是通过 toObservable() 来获取一个 Observable 对象,并通过 toBlocking() 转换为 BlockingObservable 对象,把数据以阻塞的方式反射出来,然后调用 toFuture 方法返回一个 Future,并不会阻塞,这使得消费者可以自己决定如何处理异步操作,只有 HystrixCommand 支持
-
HystrixObservableCommand对象实现了两种执行方法,如下:
- observer():返回 Observable 对象,他能操作多个结果,本质上 toObservable().subscribe(subject),可以有多个 subscriber,这个方法内部会立即订阅底层 Observable
- toObservable(): 返回 Observable 对象,他能操作多个结果,只能有一个 subscriber,这个方法内部不会订阅底层的 Observable,即除非你去订阅返回的 Observable,否则命令逻辑不会执行
-
实例化 HelloGetCommand 对象,需要创建 HystrixCommand.Setter 对象,而实例化 HelloGetObservableCommand 对象,需要创建 HystrixObservableCommand.Setter 对象,在创建对象时需要创建 HystrixCommandGroupKey 对象,用于指定命令组Key,通过使用相同命令组 Key,来统一组命令的报表、告警、仪表盘或组/库的所有权和线程池(没有单独指定);在创建对象时需要创建 HystrixThreadPoolKey 对象,用于指定来表示一个 HystrixThreadPool,以实现监控、监控指标上报、缓存和其他用途中线程池的区分。一个 HystrixCommand 与一个通过 HystrixThreadPoolKey 获得的线程池相关联,该线程池会被注入到这个命令中。如果没有指定线程池 Key,命令则默认会和一个通过 HystrixCommandGroupKey 创建的线程池相关联,示例代码如下:
HystrixCommand.Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("Hello"))
.andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("Hello"))
// HystrixObservableCommand.Setter 没有 andThreadPoolKey 方法,不能设置线程池的Key
HystrixObservableCommand.Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("HelloObservable"))
-
设置 CommandProperties 属性和 ThreadPoolProperties 属性,示例代码如下:
setter .andCommandPropertiesDefaults(HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(5000))
.andThreadPoolPropertiesDefaults(HystrixThreadPoolProperties.Setter().withMaximumSize(10));
// HystrixObservableCommand.Setter 没有线程属性设置
observableSetter.andCommandPropertiesDefaults(HystrixCommandProperties.Setter().withExecutionTimeoutInMilliseconds(3000));
-
execute() 方法调用示例代码如下:
new
HelloGetCommand(setter, restTemplate).execute(); -
queue() 方法调用示例代码如下:
Future<String> future = new
HelloGetCommand(setter,restTemplate).queue();future.get();
-
observer()方法调用示例(可以增加多个订阅)代码如下:
Observable<String> observable = hystrixObservableCommand.observe();
StringBuilder stringBuilder = new
StringBuilder();
observable.subscribe(s -> System.out.println("action1 s=" + s));
observable.subscribe(new Observer<String>() {
@Override
public
void
onCompleted() {
System.out.println("onCompleted");
}
@Override
public
void
onError(Throwable e) {
e.printStackTrace();
}
@Override
public
void
onNext(String s) {
System.out.println("onNext Thread " + Thread.currentThread().getId());
System.out.println("onNext value=" + s);
stringBuilder.append(s);
}
});
return stringBuilder.toString();
-
toObservable()方法调用(只能有一个订阅)示例如下:
Observable<String> observable = hystrixObservableCommand.toObservable();
StringBuilder stringBuilder = new
StringBuilder();
observable.subscribe(new Observer<String>() {
@Override
public
void
onCompleted() {
System.out.println("onCompleted");
}
@Override
public
void
onError(Throwable e) {
e.printStackTrace();
}
@Override
public
void
onNext(String s) {
System.out.println("onNext Thread " + Thread.currentThread().getId());
System.out.println("onNext value=" + s);
stringBuilder.append(s);
}
});
return stringBuilder.toString();