Spring Cloud系列（十四） Hystrix命令执行方式使用详解（Finchley.RC2版本）

女爷i 2022-05-15 12:11 204阅读 0赞

Hystrix命令就是我们之前说的HystrixCommand，它用来封装具体的依赖服务调用逻辑。

看一下@HystrixCommand注解的源码

/**
     * 用于标记使用Hystrix命令模式执行的方法
     */
    @Target({ElementType.METHOD})
    @Retention(RetentionPolicy.RUNTIME)
    @Inherited
    @Documented
    public @interface HystrixCommand {
    
        /**
         * 命令模式分组key，用于将如报表，告警，仪表板或团队/库所有权分组，默认值是类名
         */
        String groupKey() default "";
    
        /**
         * 命令的key值，默认值是方法名
         */
        String commandKey() default "";
    
        /**
         * 线程池key，用于表示HystrixThreadPool，用于监视，度量标准发布，缓存和其他此类用途.
         */
        String threadPoolKey() default "";
    
        /**
         * 执行降级回调方法，指定的方法必须和注解方法在同一个类中，并且必须和注解方法有相同的方法签名
         */
        String fallbackMethod() default "";
    
        /**
         * 配置命令模式的参数，具体参数对应类：HystrixCommandProperties
         */
        HystrixProperty[] commandProperties() default {};
    
        /**
         * 配置线程池参数，具体参数对应类：HystrixThreadPoolProperties
         */
        HystrixProperty[] threadPoolProperties() default {};
    
        /**
         * 定义需要忽略的异常。如果方法抛出的异常包含RUNTIME_EXCEPTION，会被包装成HystrixRuntimeException。具体逻辑在HystrixCommandAspect的hystrixRuntimeExceptionToThrowable方法
         */
        Class<? extends Throwable>[] ignoreExceptions() default {};
    
        /**
         * 定义观察者命令执行方式，详细说明见ObservableExecutionMode。判断逻辑在CommandExecutor.execute方法中
         */
        ObservableExecutionMode observableExecutionMode() default ObservableExecutionMode.EAGER;
    
        /**
         * 如果hystrix方法抛出的异常包括RUNTIME_EXCEPTION，则会被封装HystrixRuntimeException异常。此方法定义需要抛出的异常
         */
        HystrixException[] raiseHystrixExceptions() default {};
    
        /**
         * 默认降级回调方法，配置的方法不能有参数，返回值需要与注解的返回值兼容
         */
        String defaultFallback() default "";
    }

**通过继承实现命令**

我们可以通过继承的方法实现，重写run方法执行命令，并且重写getFallback方法实现降级。比如：

public class UserCommand extends HystrixCommand<User>{
    	
    	private RestTemplate restTemplate;
    	
    	private String name;
    	
    	private int age; 
    	
    	public UserCommand(RestTemplate restTemplate, String name, int age) {
    		super(HystrixCommandGroupKey.Factory.asKey("exampleGroup"));
    		this.restTemplate = restTemplate;
    		this.name = name;
    		this.age = age;
    	}
    
    	@Override
    	protected User run() throws Exception {
    		//模拟异常，测试降级策略
    		//int a = 1/0;
    		User user = restTemplate.getForObject("http://HELLO-SERVICE/hystrix/getUser?name={1}&age={2}", User.class, name,age);
    		return user;
    	}
    	/**
         * 降级。Hystrix会在run()执行过程中出现错误、超时、线程池拒绝、断路器熔断等情况时，
         * 执行getFallBack()方法内的逻辑
         */
    	@Override
    	protected User getFallback() {
    		return new User("error",0);
    	}
    
    }

通过UserCommand，我们既可以实现请求的同步执行也可以实现异步执行。

同步执行

/**
     * 测试HystrixCommand同步执行，使用继承的方式
     * @return
     */
    public String hello1(){
    	User user = new UserCommand(restTemplate, "xiaoming", 21).execute();
    	return user.toString();
    }

异步执行

/**
     * 测试HystrixCommand异步执行，使用继承的方式
     * @return
     */
    public String hello2() throws InterruptedException, ExecutionException{
    	Future<User> futureUser = new UserCommand(restTemplate, "xiaoming", 21).queue();
        //异步执行通过get获取结果
    	User user = futureUser.get();
    	return user.toString();
    }

**使用@HystrixCommand注解实现**

同步执行：

/**
     * 测试HystrixCommand同步执行，使用@HystrixCommand注解
     * @return
     */
    @HystrixCommand
    public String hello3(){
    	User user = restTemplate.getForObject("http://HELLO-SERVICE/hystrix/getUser?name={1}&age={2}", User.class, "xiaogang",12);
    	return user.toString();
    }

异步执行：

/**
     * 测试HystrixCommand异步执行，使用@HystrixCommand注解
     * @return
      */
    @HystrixCommand
    public Future<User> hello4(){
    	return new AsyncResult<User>(){
    		@Override
    		public User invoke(){
    			return restTemplate.getForObject("http://HELLO-SERVICE/hystrix/getUser?name={1}&age={2}", User.class, "xiaogang",12);
    		}
    	};
    }

除了传统的同步执行与异步执行以外，我们还可以将HystrixCommand通过Observable来实现响应式执行方式。通过调用observe()方法和toObservable()方法可以返回Observable对象，比如：

/**
     * 返回的是Hot Observable,HotObservable，不论 “事件源” 是否有“订阅者” 都会在创建后对事件进行发布。所以对于Hot
     * Observable的每一个“订阅者”都有 可能从“事件源”的中途开始的，并可能只是看到了整个操作的局部过程
     */
    public void hello5() {
    	Observable<User> hoUser = new UserCommand(restTemplate, "xiaoming21", 21).observe();
    	hoUser.subscribe(new Observer<User>() {
    		@Override
    		public void onCompleted() {
    			System.out.println("==============onCompleted");
    		}
    
    		@Override
    		public void onError(Throwable e) {
    			e.printStackTrace();
    		}
    
    		@Override
    		public void onNext(User user) {
    			System.out.println("=========onNext: " + user);
    		}
    	});
    	hoUser.subscribe(new Action1<User>() {
    		@Override
    		public void call(User user) {
    			System.out.println("==================call:" + user);
    		}
    	});
    }
    
    /**
     * Cold Observable在没有 “订阅者” 的时候并不会发布时间， 而是进行等待，知道有 “订阅者” 之后才发布事件，所以对于 Cold
     * Observable的订阅者，它可以保证从一开始看到整个操作的全部过程。
     */
    public String hello6() {
    	Observable<User> coUser = new UserCommand(restTemplate, "xiaoming22", 22).toObservable();
    	return coUser.toBlocking().single().toString();
    }

虽然HystrixCommand具备了observe()和toObservable()的功能，但是它的实现有一定的局限性，它返回的Observable只能发射一次数据，所以Hystrix还提供了HystrixObservableCommand, 通过它实现的命令可以获取能发多次的Observable。

如果使用HystrixObservableCommand来实现命令封装，需要将命令的执行逻辑在construct方法中执行，并重写 resumeWithFallback方法来实现服务降级。

public class UserObservableCommand extends HystrixObservableCommand<String> {
    
    	private RestTemplate restTemplate;
    
    	private String name;
    
    	private int age;
    
    	public UserObservableCommand(RestTemplate restTemplate, String name, int age) {
    		super(HystrixCommandGroupKey.Factory.asKey("exampleGroup"));
    		this.restTemplate = restTemplate;
    		this.name = name;
    		this.age = age;
    	}
    
    	@Override
    	protected Observable<String> construct() {
    		return Observable.create(new Observable.OnSubscribe<String>() {
                @Override
                public void call(Subscriber<? super String> subscriber) {
                    try {
                        if(!subscriber.isUnsubscribed()) {
                            //模拟异常，测试降级策略
                    		//int a = 1/0;
                        	User user = restTemplate.getForObject("http://HELLO-SERVICE/hystrix/getUser?name={1}&age={2}", User.class, name,age);
                            subscriber.onNext(user.toString());
                            subscriber.onNext(new User("测试",123).toString());
                            subscriber.onCompleted();
                        }
                    } catch (Exception e) {
                        subscriber.onError(e);
                    }
                }
            });
    	}
    
    	@Override
    	protected Observable<String> resumeWithFallback() {
    		return Observable.create(new Observable.OnSubscribe<String>() {
                @Override
                public void call(Subscriber<? super String> subscriber) {
                    try {
                        if (!subscriber.isUnsubscribed()) {
                            subscriber.onNext("失败了！");
                            subscriber.onNext("找大神来排查一下吧！");
                            subscriber.onCompleted();
                        }
                    } catch (Exception e) {
                        subscriber.onError(e);
                    }
                }
            });
    	}
    }

public void hello7() {
        Observable<String> observable= new UserObservableCommand(restTemplate,"xiaoer",1).observe();
    
        Iterator<String> iterator = observable.toBlocking().getIterator();
        while(iterator.hasNext()) {
            System.out.println(iterator.next());
        }
    }
    	
    public void hello8() {
        Observable<String> observable= new UserObservableCommand(restTemplate,"xiaoer",1).toObservable();
        Iterator<String> iterator = observable.toBlocking().getIterator();
        while(iterator.hasNext()) {
            System.out.println(iterator.next());
        }
    }

而对此的注解依然是@HystrixCommand，只是方法定义需要变化，具体内容与construct()的实现类似，如下所示：

/**
     *  EAGER参数表示使用observe()方式执行
     */
    @HystrixCommand(observableExecutionMode = ObservableExecutionMode.EAGER, fallbackMethod = "observFailed") 
    public Observable<String> hello9(final Long id) {
        return Observable.create(new Observable.OnSubscribe<String>() {
             @Override
             public void call(Subscriber<? super String> subscriber) {
                 try {
                     if(!subscriber.isUnsubscribed()) {
                         subscriber.onNext("张三的ID:");
                         int i = 1 / 0; //抛异常，模拟服务降级
                         subscriber.onNext(String.valueOf(id));
                         subscriber.onCompleted();
                     }
                 } catch (Exception e) {
                     subscriber.onError(e);
                 }
             }
        });
    }
    
    private String observFailed(Long id) {
         return "observFailed---->" + id;
    }
    
    /**
     * LAZY参数表示使用toObservable()方式执行
     */
    @HystrixCommand(observableExecutionMode = ObservableExecutionMode.LAZY, fallbackMethod = "toObserbableError") 
    public Observable<String> hello10(final String name) {
        return Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                 try {
                     if(!subscriber.isUnsubscribed()) {
                         subscriber.onNext("找到");
                         subscriber.onNext(name);
                         int i = 1/0; 抛异常，模拟服务降级
                         subscriber.onNext("了");
                         subscriber.onCompleted();
                      }
                 } catch (Exception e) {
                        subscriber.onError(e);
                 }
            }
        });
    }
    
    private String toObserbableError(String name) {
         return "toObserbableError--->" + name;
    }

@GetMapping("hello9")
    public String hello9(){
    	Iterator<String> iterator = helloService.hello9(30L).toBlocking().getIterator();
        while(iterator.hasNext()) {
             System.out.println("===============" + iterator.next());
        }
    	return "";
    }
    	
    @GetMapping("hello10")
    public String hello10(){
    	Iterator<String> iterator = helloService.hello10("张三").toBlocking().getIterator();
        while(iterator.hasNext()) {
             System.out.println("===============" + iterator.next());
        }
    	return "";
    }

在使用@HystrixCommand注解实现响应命令时，可以通过observableExecutionMode参数来控制使用observe()还是toObservable()的执行方法。

*  @HystrixCommand(observableExecutionMode = ObservableExecutionMode.EAGER)：使用observe()方式
 *  @HystrixCommand(observableExecutionMode = ObservableExecutionMode.LAZY)：使用toObservable()方式