笔记:Spring Cloud Hystrix 封装命令

使用继承的方式来创建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();

上一篇:mac怎么快速回到桌面 隐藏所有窗口


下一篇:老王带你走过 Kafka 入门教程