亿级流量电商详情页系统实战-38.hystrix开发

朴灿烈づ我的快乐病毒、 2022-11-03 04:06 262阅读 0赞

# 1. 创建Hystrix对象 #

*  HystrixCommand: 依赖服务每次返回单一的回应。
    
        HystrixCommand command = new HystrixCommand(arg1, arg2);
        
        HystrixCommand<ProductInfo> getProductInfoCommand = new GetProductInfoCommand(productId);
 *  HystrixObservableCommand: 若期望依赖服务返回一个 Observable, 并应用Observer模式监听依赖服务的回应, 用在依赖的服务返回多个操作结果的时候.
    
        HystrixObservableCommand command = new HystrixObservableCommand(arg1, arg2);
        
        HystrixObservableCommand<ProductInfo> getProductInfosCommand = 
        			new GetProductInfosCommand(productIds.split(","));
        Observable<ProductInfo> observable = getProductInfosCommand.observe();

# 2. command的四种调用方式 #

*  execute(): 同步执行, 从依赖的服务返回一个单一的结果对象, 或者是在发生错误的时候抛出异常
    
        @RequestMapping("/getProductInfo")
        @ResponseBody
        public String getProductInfo(Long productId) { 
        	HystrixCommand<ProductInfo> getProductInfoCommand = new GetProductInfoCommand(productId);
        	ProductInfo productInfo = getProductInfoCommand.execute();		
        	System.out.println(productInfo);  
        	return "success";
        }
 *  queue(): 异步执行, 直接返回一个Future对象, 其中包含了服务执行结束时要返回的单一结果对象
    
        @RequestMapping("/getProductInfo")
        	@ResponseBody
        	public String getProductInfo(Long productId) { 		
        		Future<ProductInfo> future = getProductInfoCommand.queue();
        		try { 
        			Thread.sleep(1000); 
        			System.out.println(future.get());  
        		} catch (Exception e) { 
        			e.printStackTrace();
        		}
        		return "success";
        	}
 *  observe(): 订阅一个Observable对象, Observable代表的是依赖服务返回的结果, 获取到一个那个代表结果的Observable对象的拷贝对象, 是一个 hot observable
    
        @RequestMapping("/getProductInfos")
        @ResponseBody
        public String getProductInfos(String productIds) { 
        	HystrixObservableCommand<ProductInfo> getProductInfosCommand = 
        			new GetProductInfosCommand(productIds.split(","));  
        	Observable<ProductInfo> observable = getProductInfosCommand.observe();
        	observable.subscribe(new Observer<ProductInfo>() { 
        
        		public void onCompleted() { 
        			System.out.println("获取完了所有的商品数据");
        		}
        
        		public void onError(Throwable e) { 
        			e.printStackTrace();
        		}
        
        		public void onNext(ProductInfo productInfo) { 
        			System.out.println(productInfo);  
        		}
        		
        	});
        	return "success";
        }
 *  toObservable(): 返回一个Observable对象, 如果我们订阅这个对象, 就会执行command并且获取返回结果, 是一个 cold observable
    
        @RequestMapping("/getProductInfos")
        @ResponseBody
        public String getProductInfos(String productIds) { 
        		HystrixObservableCommand<ProductInfo> getProductInfosCommand = 
        				new GetProductInfosCommand(productIds.split(","));  
        	
        		observable = getProductInfosCommand.toObservable(); // 还没有执行
        		
        		observable.subscribe(new Observer<ProductInfo>() {  // 等到调用subscribe然后才会执行
        
        			public void onCompleted() { 
        				System.out.println("获取完了所有的商品数据");
        			}
        
        			public void onError(Throwable e) { 
        				e.printStackTrace();
        			}
        
        			public void onNext(ProductInfo productInfo) { 
        				System.out.println(productInfo);  
        			}
        			
        		});
        		return "success";
        	}

execute()实际上会调用queue().get().queue()，接着会调用toObservable().toBlocking().toFuture()。也就是说, 无论是哪种执行command的方式, 最终都是依赖toObservable()去执行。

# 3. 检查是否开启缓存 #

若当前命令的请求缓存功能(request cache)是被启用的, 并且该命令缓存命中, 那么缓存的结果会立即以Observable对象的形式返回。

# 4.检查是否开启了短路器 #

在命令结果没有被缓存命中的时候, Hystrix会在执行命令前检查断路器是否为打开状态：

*  如果断路器是打开的, 那么Hystrix不会执行命令, 而是转接到fallback处理逻辑（第\#8）
 *  如果断路器是关闭的, Hystrix会检查是否有可用资源来执行命令（第\#5）

# 5.检查线程池/队列/semaphore是否已经满了 #

如果command对应的线程池, 列队, semeaphore已经满了, 那么也不会执行command, 而是直接调用fallback降级机制。

# 6.执行command #

Hystrix会根据我们编写的方法来决定采取什么样的方式去请求依赖服务：

*  HystrixCommand.run()：返回单个响应或抛出异常。
    
        public class GetProductInfoCommand extends HystrixCommand<ProductInfo> { 
        
        	private Long productId;
        	
        	public GetProductInfoCommand(Long productId) { 
        		super(HystrixCommandGroupKey.Factory.asKey("GetProductInfoGroup"));
        		this.productId = productId;
        	}
        	
        	@Override
        	protected ProductInfo run() throws Exception { 
        		String url = "http://127.0.0.1:8082/getProductInfo?productId=" + productId;
        		String response = HttpClientUtils.sendGetRequest(url);
        		return JSONObject.parseObject(response, ProductInfo.class);  
        	}
        
        }
 *  HystrixObservableCommand.construct()：返回Observable对象来发射多个结果, 或通过onError发送错误通知。
    
        public class GetProductInfosCommand extends HystrixObservableCommand<ProductInfo> { 
        
        	private String[] productIds;
        	
        	public GetProductInfosCommand(String[] productIds) { 
        		super(HystrixCommandGroupKey.Factory.asKey("GetProductInfoGroup"));
        		this.productIds = productIds;
        	}
        	
        	@Override
        	protected Observable<ProductInfo> construct() { 
        		return Observable.create(new Observable.OnSubscribe<ProductInfo>() { 
        
        			public void call(Subscriber<? super ProductInfo> observer) { 
        				try { 
        					for(String productId : productIds) { 
        						String url = "http://127.0.0.1:8082/getProductInfo?productId=" + productId;
        						String response = HttpClientUtils.sendGetRequest(url);
        						ProductInfo productInfo = JSONObject.parseObject(response, ProductInfo.class); 
        						observer.onNext(productInfo); 
        					}
        					observer.onCompleted();
        				} catch (Exception e) { 
        					observer.onError(e);  
        				}
        			}
        			
        		}).subscribeOn(Schedulers.io());
        	}
        
        }
 *  如果HystrixCommand.run()或HystrixObservableCommand.construct()的执行, 超过了timeout时长的话, 那么command所在的线程就会抛出一个TimeoutException。  
    (1) 如果timeout了, 也会去执行fallback降级机制, 而且就不会管run()或construct()返回的值了。  
    (2) 如果没有timeout的话, 那么就会拿到一些调用依赖服务获取到的结果, 然后hystrix会做一些logging记录和metric统计。

# 7. 短路健康检查 #

*  Hystrix会将每一个依赖服务的调用成功, 失败, 拒绝, 超时等事件, 都会发送给circuit breaker断路器。
 *  短路器就会对调用成功/失败/拒绝/超时等事件的次数进行统计。
 *  短路器会根据这些统计次数来决定, 是否要进行短路, 如果打开了短路器, 那么在一段时间内就会直接短路, 然后如果在之后第一次检查发现调用成功了, 就关闭断路器。

# 8. fallback处理 #

当命令执行失败时, Hystrix会进入fallback尝试回退处理, 也叫服务降级, 可以引入服务降级的请求有下面几种:

*  run()或construct()抛出一个异常
 *  短路器打开
 *  线程池/队列/semaphore满了
 *  command执行超时了

一般在降级机制中, 都建议给出一些默认的返回值, 比如静态的一些代码逻辑, 或者从内存中的缓存中提取一些数据, 尽量在这里不要再进行网络请求了。即使在降级中, 一定要进行网络调用, 也应该将那个调用放在一个HystrixCommand中，进行隔离.

*  在HystrixCommand中，上线getFallback()方法, 可以提供降级机制。
 *  在HystirxObservableCommand中, 实现一个resumeWithFallback()方法, 返回一个Observable对象, 可以提供降级结果。

不同的command执行方式，其fallback为空或者异常时的返回结果不同:

*  execute()：抛出异常
 *  queue()：成功时返回java.util.concurrent.Future, 但如果调用 Future.get()将抛出异常
 *  observe()：返回 Observable 对象, 当你订阅该 Observable 时, 将会立即终止并且调用订阅者的onError方法。
 *  toObservable()：同observe()

# 9. 返回成功的响应 #

当Hystrix命令执行成功后, 会将处理结果直接返回或是以Observable的形式返回, 具体怎么返回要根据执行命令的方式来区分。

*  execute(), 获取一个Future.get(), 然后拿到单个结果。
 *  queue(), 返回一个Future。
 *  observer(), 立即订阅Observable, Hystrix会回调结果。
 *  toObservable(), 返回一个原始的Observable, 必须手动订阅才会去执行。