Spring Cloud Ribbon && Hystrix

妖狐艹你老母 2022-05-25 03:05 98阅读 0赞

## Ribbon 是实现客户端负载均衡的 ##

<?xml version="1.0" encoding="UTF-8"?>
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 <modelVersion>4.0.0</modelVersion>
 
 <groupId>com.example</groupId>
 <artifactId>service-ribbon</artifactId>
 <version>0.0.1-SNAPSHOT</version>
 <packaging>jar</packaging>
 
 <name>service-ribbon</name>
 <description>Demo project for Spring Cloud</description>
 
 <parent>
 <groupId>org.springframework.boot</groupId>
 <artifactId>spring-boot-starter-parent</artifactId>
 <version>1.5.12.RELEASE</version>
 <relativePath/> 
 </parent>
 
 <properties>
 <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 <java.version>1.8</java.version>
 <spring-cloud.version>Edgware.SR3</spring-cloud.version>
 </properties>
 
 <dependencies>
 <dependency>
 <groupId>org.springframework.boot</groupId>
 <artifactId>spring-boot-starter-web</artifactId>
 </dependency>
 <dependency>
 <groupId>org.springframework.cloud</groupId>
 <artifactId>spring-cloud-starter-eureka</artifactId>
 </dependency>
 <dependency>
 <groupId>org.springframework.cloud</groupId>
 <artifactId>spring-cloud-starter-ribbon</artifactId>
 </dependency>
 
 <dependency>
 <groupId>org.springframework.cloud</groupId>
 <artifactId>spring-cloud-starter-hystrix</artifactId>
 </dependency>
 
 
 <dependency>
 <groupId>org.springframework.boot</groupId>
 <artifactId>spring-boot-starter-actuator</artifactId>
 </dependency>
 <dependency>
 <groupId>org.springframework.cloud</groupId>
 <artifactId>spring-cloud-starter-hystrix-dashboard</artifactId>
 </dependency>
 
 <dependency>
 <groupId>org.springframework.boot</groupId>
 <artifactId>spring-boot-starter-test</artifactId>
 <scope>test</scope>
 </dependency>
 </dependencies>
 
 <dependencyManagement>
 <dependencies>
 <dependency>
 <groupId>org.springframework.cloud</groupId>
 <artifactId>spring-cloud-dependencies</artifactId>
 <version>${spring-cloud.version}</version>
 <type>pom</type>
 <scope>import</scope>
 </dependency>
 </dependencies>
 </dependencyManagement>
 
 <build>
 <plugins>
 <plugin>
 <groupId>org.springframework.boot</groupId>
 <artifactId>spring-boot-maven-plugin</artifactId>
 </plugin>
 </plugins>
 </build>
 
 
 </project>

application.ym

server:
      port: 8084
      context-path: /service-ribbon
    eureka:
      client:
        service-url:
          defaultZone: http://localhost:8081/eureka-server/eureka/
      instance:
    # 使用 spring boot admin 时,如果设置了 server.context-path,需要配置下面2个,否则在admin server 无法获取完整的监控数据
        metadata-map:
          management.context-path: ${server.context-path}
          management.context-port: ${server.port}
    # spring.application.name,这个很重要，在以后的服务与服务之间相互调用一般都是根据这个name
    spring:
      application:
        name: service-ribbon

关闭actuator监控的权限验证  
`management.security.enabled=false`

**Ribbon的负载均衡底层是通过对RestTemplate添加拦截器实现的,必须要配置RestTemplate的bean,并加上 @LoadBalanced 注解**

@EnableDiscoveryClient
 @EnableHystrix
 @EnableHystrixDashboard
 @ServletComponentScan
 @SpringBootApplication
 public class ServiceRibbonApplication { 
 
 public static void main(String[] args) {
 SpringApplication.run(ServiceRibbonApplication.class, args);
 }
 
 /** * 负载均衡的策略可以查看IRule的实现类 * * 具体的配置: * 1. <clientName>.ribbon.NFLoadBalancerRuleClassName=com.netflix.loadbalancer.RandomRule * 2. @Bean * public IRule ribbonRule() { * return new RandomRule();//这里配置策略，和配置文件对应 * } */
 @Bean
 @LoadBalanced
 RestTemplate restTemplate() {
 return new RestTemplate();
 }
 
 }

## Hystrix 是熔断器，可以实现服务降级 ##

HelloControler.java

@RestController
    public class HelloControler { 
        @Autowired
        private HelloService helloService;
    
        @RequestMapping("/hi/{name}")
        public String hi(@PathVariable("name") String name) {
            helloService.hiService(name);
            return helloService.hiService(name);
        }
    
        @RequestMapping("/syn/{name}")
        public String synCommand(@PathVariable("name") String name) {
            helloService.synCommand(name);
            return helloService.synCommand(name);
        }
    
        @RequestMapping("/asyn/{name}")
        public String asynCommand(@PathVariable("name") String name) {
            helloService.asynCommand(name);
            return helloService.asynCommand(name);
        }
    
        @RequestMapping("/test")
        public void testCollapser() throws ExecutionException, InterruptedException {
            helloService.testCollapser();
        }
    }

HelloService.java

@Service
 public class HelloService { 
 
 private static final Logger logger = LoggerFactory.getLogger(HelloService.class);
 
 @Autowired
 private RestTemplate restTemplate;
 
 /** * @HystrixCommand 注解对该方法创建了熔断器的功能,并指定了fallbackMethod熔断方法. * 如果调用的服务出现故障,或者调用的服务超时(Hystrix 默认超时时间为2000ms)会触发熔断请求. * * 出现异常时,除了HystrixBadRequestException之外,其他异常均会触发服务降级的处理逻辑. * 可以通过 @HystrixCommand 注解的 ignoreExceptions 参数指定不需要触发服务降级的异常. * * 使用注解实现服务降级时,@HystrixCommand和fallback必须定义在同一个类中,并且 fallbackMethod 的值必须和 fallback 方法的名字相同. * * 不使用注解,也可以通过自定义类继承 HystrixCommand, 并重写run 和 getFallback 方法实现服务降级 * * 获取具体的异常: * 1.自定义类,在 getFallback() 方法中直接使用 getExecutionException() 即可得到 * 2.注解,在 fallbackMethod 指定的方法参数中添加 Throwable e * * 开启请求缓存: * 1.自定义类中重写 getCacheKey() 方法,Hystrix会根据方法返回的值来区分是否是重复的请求.如果cacheKey相同,只会在第一个请求到达时真正的调用一次,后面的请求直接从缓存中返回结果. * * 2.通过JSR 的注解( @CacheResult,@CacheKey 和 @CacheRemove) * 可以方法上使用 @CacheResult 注解,表示给该方法开启缓存,默认情况下方法的所有参数都将作为缓存的key(必须和 @HystrixCommand 注解配合使用) * 可以使用 @CacheKey 来指定缓存的key(如果使用了 @CacheResult 和 @CacheRemove 中的 cacheKeyMethod 参数,则 @CacheKey 不会起作用) * 使用 @CacheRemove 注解时,commandKey 参数必须指定(只有配置了该参数,Hystrix才能找到正确的请求命令缓存位置) * * 还需要通过servlet的Filter配置hystrix的上下文(HystrixRequestContext.initializeContext();),否则会报错(Request caching is not available. Maybe you need to initialize the HystrixRequestContext?). * 这个缓存是基于request的(Key to be used for request caching).在不同context中的缓存是不共享的,还有这个request内部一个ThreadLocal,所以request只能限于当前线程. * (每请求一次controller就会走一次filter,上下文又会初始化一次,前面缓存的就失效了,又得重新来.所以要是想测试缓存,得在一次controller请求中多次调用那个加了缓存的service或HystrixCommand命令...) * * Hystrix中的请求合并: * 利用一个合并处理器,将对同一个服务发起的连续请求合并成一个请求进行处理(这些连续请求的时间窗默认为10ms),在这个过程中涉及到的一个核心类就是HystrixCollapser.自定义类 MyCollapser 继承 HystrixCollapser * 多个请求被合并为一个请求进行一次性处理,可以有效节省网络带宽和线程池资源.但是,有优点必然也有缺点, * 设置请求合并之后,本来一个请求可能5ms就搞定了,但是现在必须再等10ms看看还有没有其他的请求一起的,这样一个请求的耗时就从5ms增加到15ms了. * 不过,如果我们要发起的命令本身就是一个高延迟的命令,那么这个时候就可以使用请求合并了,因为这个时候时间窗的时间消耗就显得微不足道了,另外高并发也是请求合并的一个非常重要的场景。 */
 @CacheResult
 @HystrixCommand(fallbackMethod = "hiError")
 public String hiService(@CacheKey String name) {
 logger.info("enter the method hiService...");
 return restTemplate.getForObject("http://EUREKA-CLIENT/eureka-client/hi/" + name, String.class);
 }
 
 public String hiError(String name, Throwable e) {
 logger.error(e.getMessage());
 return "hi " + name + ",error! (by annotation)";
 }
 
 
 /** * 同步执行 * * @param name * @return */
 public String synCommand(String name) {
 /** * Setter 实例只能通过 withGroupKey() 的静态方法来创建,GroupKey 是必须的,而CommandKey 是可选的. * * 必须设置命名组的原因: * 1.根据命名组实现统计 * 2.Hystrix 命令默认的线程划分是根据命名组来实现的(默认情况下,Hystrix会让相同组名的命令使用同一个线程池) */
 com.netflix.hystrix.HystrixCommand.Setter setter = com.netflix.hystrix.HystrixCommand.Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("MyCommand"))
 .andCommandKey(HystrixCommandKey.Factory.asKey("synCommand"));
 
 MyCommand command = new MyCommand(setter, restTemplate, name);
 
 return command.execute();
 }
 
 /** * 异步执行 * * @param name * @return */
 public String asynCommand(String name) {
 com.netflix.hystrix.HystrixCommand.Setter setter = com.netflix.hystrix.HystrixCommand.Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("MyCommand"))
 .andCommandKey(HystrixCommandKey.Factory.asKey("asynCommand"));
 
 MyCommand command = new MyCommand(setter, restTemplate, name);
 
 Future<String> future = command.queue();
 
 String result = null;
 try {
 result = future.get();
 } catch (Exception e) {
 logger.error("异步执行失败...", e);
 }
 return result;
 }
 
 public List<String> hiAll(List<String> names) {
 return restTemplate.getForObject("http://EUREKA-CLIENT/eureka-client/hiAll/" + StringUtils.join(names, ","), List.class);
 }
 
 /** * 先发送3个请求,然后睡眠1秒钟,再发起1个请求,这样,前3个请求就会被合并为一个请求,第四个请求因为间隔的时间比较久,所以不会被合并,而是单独创建一个线程去处理。 */
 public void testCollapser() throws ExecutionException, InterruptedException {
 MyCollapser bc1 = new MyCollapser("a", this);
 MyCollapser bc2 = new MyCollapser("b", this);
 MyCollapser bc3 = new MyCollapser("c", this);
 MyCollapser bc4 = new MyCollapser("d", this);
 Future<String> q1 = bc1.queue();
 Future<String> q2 = bc2.queue();
 Future<String> q3 = bc3.queue();
 String result1 = q1.get();
 String result2 = q2.get();
 String result3 = q3.get();
 Thread.sleep(1000);
 Future<String> q4 = bc4.queue();
 String result4 = q4.get();
 System.out.println("1>>>" + result1);
 System.out.println("2>>>" + result2);
 System.out.println("3>>>" + result3);
 System.out.println("4>>>" + result4);
 }
 }

**不使用@HystrixCommand 注解，也可以通过继承HystrixCommand来实现**

MyCommand.java

public class MyCommand extends HystrixCommand<String> { 
 
 private static final Logger logger = LoggerFactory.getLogger(MyCommand.class);
 
 private String name;
 private RestTemplate restTemplate;
 
 public MyCommand(Setter setter, RestTemplate restTemplate, String name) {
 super(setter);
 this.restTemplate = restTemplate;
 this.name = name;
 }
 
 @Override
 protected String run() throws Exception {
 logger.info("enter the MyCommand run() ...");
 return restTemplate.getForObject("http://EUREKA-CLIENT/eureka-client/hi/" + name, String.class);
 }
 
 @Override
 protected String getFallback() {
 logger.error(getExecutionException().getMessage());
 return "hi " + name + ",error! (by extend HystrixCommand)";
 }
 
 // 开启缓存
 @Override
 protected String getCacheKey() {
 return String.valueOf(name);
 }
 }

**使用Hystrix缓存需要在Filter中配置hystrix的上下文**

HystrixRequestContextServletFilter.java

//在启动类上添加 @ServletComponentScan 注解
    @WebFilter(filterName = "hystrixRequestContextServletFilter", urlPatterns = "/*", asyncSupported = true)
    public class HystrixRequestContextServletFilter implements Filter { 
    
        @Override
        public void init(FilterConfig filterConfig) throws ServletException {
    
        }
    
        @Override
        public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
        // 初始化hystrix的上下文
            HystrixRequestContext context = HystrixRequestContext.initializeContext();
            try {
                chain.doFilter(request, response);
            } finally {
                context.shutdown();
            }
        }
    
        @Override
        public void destroy() {
    
        }
    }

**Hystrix中的请求合并: 继承HystrixCollapser**

MyCollapser.java

public class MyCollapser extends HystrixCollapser<List<String>, String, String> { 
 
 private String name;
 
 private HelloService service;
 
 /** * 1.首先在构造方法中,设置了请求时间窗为100ms,即请求时间间隔在100ms之内的请求会被合并为一个请求。 * 2.createCommand方法主要用来合并请求,在这里获取到各个单个请求的name，将这些单个的name放到一个集合中,然后再创建出一个MyBatchCommand对象，用该对象去发起一个批量请求。 * 3.mapResponseToRequests方法主要用来为每个请求设置请求结果.该方法的第一个参数batchResponse表示批处理请求的结果,第二个参数collapsedRequests则代表了每一个被合并的请求,然后我们通过遍历batchResponse来为collapsedRequests设置请求结果。 */
 public MyCollapser(String name, HelloService service) {
 super(Setter.withCollapserKey(HystrixCollapserKey.Factory.asKey("myCollapser"))
 .andCollapserPropertiesDefaults(HystrixCollapserProperties.Setter().
 withTimerDelayInMilliseconds(100)));
 this.name = name;
 this.service = service;
 }
 
 @Override
 public String getRequestArgument() {
 return name;
 }
 
 @Override
 protected HystrixCommand<List<String>> createCommand(Collection<CollapsedRequest<String, String>> collapsedRequests) {
 ArrayList<String> names = new ArrayList<>(collapsedRequests.size());
 names.addAll(collapsedRequests.stream().map(CollapsedRequest::getArgument).collect(Collectors.toList()));
 return new MyBatchCommand(service, names);
 }
 
 @Override
 protected void mapResponseToRequests(List<String> batchResponse, Collection<CollapsedRequest<String, String>> collapsedRequests) {
 int count = 0;
 for (CollapsedRequest<String, String> request : collapsedRequests) {
 String result = batchResponse.get(count++);
 request.setResponse(result);
 }
 }
 }

MyBatchCommand.java

public class MyBatchCommand extends HystrixCommand<List<String>> { 
 
 private HelloService service;
 
 private List<String> names;
 
 protected MyBatchCommand(HelloService service, List<String> names) {
 super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("MyCommand")));
 this.service = service;
 this.names = names;
 }
 
 @Override
 protected List<String> run() throws Exception {
 return service.hiAll(names);
 }
 
 @Override
 protected List<String> getFallback() {
 return Arrays.asList(new String[]{
 "MyBatchCommand error!"});
 }
 }

\*\*添加 @EnableHystrixDashboard 注解开启 Hystrix 仪表盘功能  
输入 [http://hostname:port/context-path/hystrix][http_hostname_port_context-path_hystrix] 进入 Hystrix Dashboard 页面\*\*

![这里写图片描述][70]

点击 Monitor Stream

![这里写图片描述][70 1]

有时候可能会出现 Unable to connect to Command Metric Stream.这个错误.

添加HystrixMetricsStreamServlet

@Bean(name = "hystrixRegistrationBean")
        public ServletRegistrationBean servletRegistrationBean() {
            ServletRegistrationBean registration = new ServletRegistrationBean(
                    new HystrixMetricsStreamServlet(), "/hystrix.stream");
            registration.setName("HystrixMetricsStreamServlet");
            registration.setLoadOnStartup(1);
            return registration;
        }

[http_hostname_port_context-path_hystrix]: http://hostname:port/context-path/hystrix
[70]: /images/20220525/3b0457cb42164165a0c1a48d7cb15804.png
[70 1]: /images/20220525/9d34f6f4ad374954b75425c782573441.png