欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

Spring Cloud服务熔断

程序员文章站 2024-03-20 23:15:34
...

版本信息

Spring Cloud : Hoxton.SR1

Spring Boot : 2.2.2.RELEASE

Spring Cloud Hystrix使用

熔断策略

THREAD — it executes on a separate thread and concurrent requests are
limited by the number of threads in the thread-pool

SEMAPHORE — it executes on the calling thread and concurrent requests are limited by the semaphore count

  • 线程方式 超时,基于 Thread Pool + Future实现
  • 信号量方式 限流,基于Semaphore实现

基本使用
1.添加pom依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>

<!--hystrix-->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-netflix-hystrix</artifactId>
</dependency>

2.服务启动类添加@EnableCircuitBreaker

/**
 * 服务熔断应用
 *
 * @author FelixFly <[email protected]>
 * @date 2020/2/5
 */
@SpringBootApplication
@EnableCircuitBreaker
public class CircuitBreakerApplication {

    public static void main(String[] args) {
        SpringApplication.run(CircuitBreakerApplication.class, args);
    }
}

3.注解方式演示服务端点

/**
 * 注解演示服务端点
 *
 * @author FelixFly <[email protected]>
 * @date 2020/2/5
 */
@RestController
@RequestMapping("/annotation")
public class AnnotationDemoController {

    @HystrixCommand(fallbackMethod = "fallBack", commandProperties = {
            @HystrixProperty(name = "execution.isolation.strategy", value = "THREAD"),
            @HystrixProperty(name = "execution.isolation.thread.timeoutInMilliseconds", value = "50")
    })
    @GetMapping("/thread/{message}")
    public String thread(@PathVariable String message) {
        await();
        return "Thread:" + message;
    }

    @HystrixCommand(fallbackMethod = "fallBack", commandProperties = {
            @HystrixProperty(name = "execution.isolation.strategy", value = "SEMAPHORE"),
            @HystrixProperty(name = "execution.isolation.thread.timeoutInMilliseconds", value = "50"),
            @HystrixProperty(name = "execution.isolation.semaphore.maxConcurrentRequests", value = "1")
    })
    @GetMapping("/semaphore/{message}")
    public String hello(@PathVariable String message) {
        await();
        return "Semaphore:" + message;
    }

    public String fallBack(String message) {
        return "Fallback:" + message;
    }

    public void await() {
        int waitTime = new Random().nextInt(100);
        try {
            System.out.printf("线程【%s】执行时间%d毫秒\n", Thread.currentThread().getName(), waitTime);
            TimeUnit.MILLISECONDS.sleep(waitTime);
        } catch (InterruptedException e) {
            throw new RuntimeException("线程中断");
        }
    }

}

@HystrixCommand配置属性:https://github.com/Netflix/Hystrix/wiki/Configuration#CommandProperties

4.配置文件application.yml

server:
  port: 8090

Thread方式访问地址:http://127.0.0.1:8090/annotation/thread/123会出现两种结果

  • Thread:123 正常返回结果
  • Fallback:123 超时返回结果

控制台打印

线程【hystrix-DemoController-6】执行时间59毫秒
线程【hystrix-DemoController-7】执行时间12毫秒
线程【hystrix-DemoController-8】执行时间44毫秒
线程【hystrix-DemoController-9】执行时间22毫秒
线程【hystrix-DemoController-10】执行时间58毫秒

从控制台来看,这个切换线程了

Semaphore访问地址:http://127.0.0.1:8090/annotation/semaphore/123会出现两种结果

  • Semaphore:123 正常返回结果
  • Fallback:123 超时返回结果

控制台打印

线程【http-nio-8090-exec-1】执行时间0毫秒 线程【http-nio-8090-exec-2】执行时间50毫秒
线程【http-nio-8090-exec-3】执行时间53毫秒 线程【http-nio-8090-exec-4】执行时间1毫秒
线程【http-nio-8090-exec-5】执行时间35毫秒 线程【http-nio-8090-exec-6】执行时间91毫秒
线程【http-nio-8090-exec-7】执行时间5毫秒 线程【http-nio-8090-exec-8】执行时间68毫秒

从控制台来看,这个没有切换线程

基于Spring Aop自定义服务熔断

服务熔断策略枚举

/**
 * 服务熔断策略
 *
 * @author FelixFly <[email protected]>
 * @date 2020/2/5
 */
public enum CircuitBreakerStrategy {
    /**
     * 超时
     */
    THREAD,
    /**
     * 限流
     */
    SEMAPHORE;
}

服务熔断注解

/**
 * 服务熔断注解
 *
 * @author FelixFly <[email protected]>
 * @date 2020/2/5
 */
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface CircuitBreakerCommand {

    /**
     * 服务熔断策略,默认为{@link CircuitBreakerStrategy#SEMAPHORE}
     */
    CircuitBreakerStrategy strategy() default CircuitBreakerStrategy.SEMAPHORE;

    /**
     * 信号量
     */
    int semaphore() default 10;

    /**
     * 超时时间
     */
    int timeout() default 50;

    /**
     * 超时时间单位
     */
    TimeUnit timeUnit() default TimeUnit.MILLISECONDS;

    /**
     * 回调方法
     */
    String fallBack() default "";
}

服务熔断器(采用策略+模板方法模式)

  • 服务熔断器接口
/**
 * 服务熔断器
 *  * @author FelixFly <[email protected]>
 * @date 2020/2/5
 */
public interface CircuitBreaker {

    /**
     * 支持的策略
     *
     * @param strategy 策略
     * @return true 支持  false 不支持
     */
    boolean isSupport(CircuitBreakerStrategy strategy);


    /**
     * 执行方法
     *
     * @param joinPoint      切面执行
     * @param method         执行方法
     * @param circuitBreakerCommand 注解信息
     * @return 执行结果
     */
    Object execute(ProceedingJoinPoint joinPoint, Method method, CircuitBreakerCommand circuitBreakerCommand) throws Exception;
}
  • 服务熔断器抽象实现类
/**
 * 服务熔断器抽象实现
 *  * @author FelixFly <[email protected]>
 * @date 2020/2/5
 */
public abstract class AbstractCircuitBreaker implements CircuitBreaker {

    @Override
    public Object execute(ProceedingJoinPoint joinPoint, Method method, CircuitBreakerCommand circuitBreakerCommand)
            throws Exception {
        // 目标对象
        Object target = joinPoint.getTarget();
        // 执行方法参数
        Object[] args = joinPoint.getArgs();
        try {
            return internalExecute(method, joinPoint, circuitBreakerCommand);
        } catch (TimeoutException exception) {
            // 对TimeoutException异常进行回调方法
            String fallBack = circuitBreakerCommand.fallBack();
            if (StringUtils.isNotBlank(fallBack)) {
                return invokeFallBackMethod(method, target, circuitBreakerCommand, args);
            }
            throw exception;
        }
    }


    /**
     * 内部执行方法
     *
     * @param method         执行方法
     * @param joinPoint         切面执行
     * @param circuitBreakerCommand 服务熔断注解信息
     * @return 执行结果
     * @throws TimeoutException 超时异常
     */
    public abstract Object internalExecute(Method method, ProceedingJoinPoint joinPoint,
                                           CircuitBreakerCommand circuitBreakerCommand)
            throws Exception;

    /**
     * 执行服务熔断方法
     *
     * @param method         执行方法
     * @param target         目前对象
     * @param circuitBreakerCommand 服务熔断注解信息
     * @param args           方法参数
     * @return 执行结果
     */
    protected Object invokeFallBackMethod(Method method, Object target,
                                          CircuitBreakerCommand circuitBreakerCommand, Object[] args) throws Exception {
        Method fallBackMethod = fallBackMethod(method, target, circuitBreakerCommand);
        return fallBackMethod.invoke(target, args);
    }


    /**
     * 获取回调方法
     *
     * @param method         执行方法
     * @param target         目前对象
     * @param circuitBreakerCommand 服务熔断注解信息
     * @return 目标方法
     * @throws NoSuchMethodException 无此目标方法
     */
    private Method fallBackMethod(Method method, Object target,
                                  CircuitBreakerCommand circuitBreakerCommand) throws NoSuchMethodException {
        String fallBackMethodName = circuitBreakerCommand.fallBack();
        Class<?> beanClass = target.getClass();
        return beanClass.getMethod(fallBackMethodName, method.getParameterTypes());
    }
}
  • 超时熔断器
/**
 * {@link CircuitBreakerStrategy#THREAD} 超时熔断器
 *  * @author FelixFly <[email protected]>
 * @date 2020/2/5
 */
@Service
public class ThreadCircuitBreaker extends AbstractCircuitBreaker {

    private ExecutorService executorService = Executors.newFixedThreadPool(2);

    @Override
    public boolean isSupport(CircuitBreakerStrategy strategy) {
        return CircuitBreakerStrategy.THREAD.equals(strategy);
    }

    @Override
    public Object internalExecute(Method method, ProceedingJoinPoint joinPoint,
                                  CircuitBreakerCommand circuitBreakerCommand) throws Exception {
        // 执行方法参数
        Object[] args = joinPoint.getArgs();

        // 进行异步调用
        Future<Object> future = executorService.submit(() -> {
            try {
                return joinPoint.proceed(args);
            } catch (Throwable throwable) {
                throw new RuntimeException("调用目标方法错误");
            }
        });
        try {
            return future.get(circuitBreakerCommand.timeout(), circuitBreakerCommand.timeUnit());
        } catch (TimeoutException timeoutException) {
            future.cancel(true);//取消执行
            throw timeoutException;
        }

        // 使用CompletableFuture进行异步调用,超时异常没有办法取消调用
        /*CompletableFuture<Object> completableFuture = CompletableFuture.supplyAsync(() -> {

            try {
                return joinPoint.proceed(args);
            } catch (Throwable throwable) {
                throw new RuntimeException("调用目标方法错误");
            }
        });
        try {
            return completableFuture.get(circuitBreakerCommand.timeout(), circuitBreakerCommand.timeUnit());
        } catch (TimeoutException timeoutException) {
            // this value has no effect in this implementation because interrupts are not used to control processing.
            completableFuture.cancel(true);//取消执行,无法进行取消
            throw timeoutException;
        }*/
    }
    
    @PreDestroy
    public void destroy(){
        executorService.shutdown();
    }
}
  • 限流熔断器
/**
 * {@link CircuitBreakerStrategy#SEMAPHORE} 限流熔断器
 *
 * @author FelixFly <[email protected]>
 * @date 2020/2/5
 */
@Service
public class SemaphoreCircuitBreaker extends AbstractCircuitBreaker {

    private Map<Method, Semaphore> methodSemaphoreMap = new ConcurrentHashMap<>(128);

    @Override
    public boolean isSupport(CircuitBreakerStrategy strategy) {
        return CircuitBreakerStrategy.SEMAPHORE.equals(strategy);
    }

    @Override
    public Object internalExecute(Method method, ProceedingJoinPoint joinPoint,
                                  CircuitBreakerCommand circuitBreakerCommand) throws Exception {
        Semaphore semaphore = initSemaphore(method, circuitBreakerCommand.semaphore());
        boolean tryAcquire = semaphore.tryAcquire(circuitBreakerCommand.timeout(), circuitBreakerCommand.timeUnit());
        if (!tryAcquire) {
            throw new TimeoutException("获取信号量超时");
        }
        try {
            return joinPoint.proceed(joinPoint.getArgs());
        } catch (Throwable throwable) {
            throw new RuntimeException("调用目标方法错误");
        } finally {
            semaphore.release();
        }
    }

    private Semaphore initSemaphore(Method method, int semaphore) {
        return methodSemaphoreMap.computeIfAbsent(method,
                value -> new Semaphore(semaphore));
    }
}

服务熔断切面

/**
 * {@link CircuitBreakerCommand} 服务熔断切面
 *
 * @author FelixFly <[email protected]>
 * @date 2020/2/5
 */
@Configuration
@Aspect
public class CircuitBreakerAop {

    @Autowired
    private List<CircuitBreaker> circuitBreakers;

    @Around("@annotation(top.felixfly.cloud.circuit.breaker.annotation.CircuitBreakerCommand)")
    public Object circuitBreaker(ProceedingJoinPoint joinPoint) throws Exception {
        // @CircuitBreaker直接注解在方法上,肯定是直接返回
        MethodSignature methodSignature = (MethodSignature) joinPoint.getSignature();
        // 执行方法
        Method method = methodSignature.getMethod();
        CircuitBreakerCommand circuitBreakerCommand = method.getAnnotation(CircuitBreakerCommand.class);

        CircuitBreakerStrategy strategy = circuitBreakerCommand.strategy();
        CircuitBreaker circuitBreaker = circuitBreakers.stream().filter(item -> item.isSupport(strategy))
                .findAny().orElseThrow(() -> new RuntimeException("无此服务策略[" + strategy + "]"));
        return circuitBreaker.execute(joinPoint,method, circuitBreakerCommand);
    }
}

自定义服务端点

/**
 * 自定义演示服务端点
 *
 * @author FelixFly <[email protected]>
 * @date 2020/2/5
 */
@RestController
@RequestMapping("custom")
public class CustomDemoController {

    @CircuitBreakerCommand(strategy = CircuitBreakerStrategy.THREAD, fallBack = "fallBack")
    @GetMapping("/thread/{message}")
    public String thread(@PathVariable String message) {
        await();
        System.out.println("执行Thread");
        return "Thread:" + message;
    }

    @CircuitBreakerCommand(strategy = CircuitBreakerStrategy.SEMAPHORE, semaphore = 1, fallBack = "fallBack")
    @GetMapping("/semaphore/{message}")
    public String hello(@PathVariable String message) {
        await();
        System.out.println("执行Semaphore");
        return "Semaphore:" + message;
    }

    public String fallBack(String message) {
        return "Fallback:" + message;
    }

    public void await() {
        // 100的时候限流熔断不太好演示,改成300进行演示
        int waitTime = new Random().nextInt(300);
        try {
            System.out.printf("线程【%s】执行时间%d毫秒\n", Thread.currentThread().getName(), waitTime);
            TimeUnit.MILLISECONDS.sleep(waitTime);
        } catch (InterruptedException e) {
            throw new RuntimeException("线程中断");
        }
    }

}

Thread方式访问地址:http://127.0.0.1:8090/custom/thread/123会出现两种结果

  • Thread:123 正常返回结果
  • Fallback:123 超时返回结果

Semaphore访问地址:http://127.0.0.1:8090/custom/semaphore/123会出现两种结果

  • Semaphore:123 正常返回结果
  • Fallback:123 超时返回结果

Spring Cloud Hystrix源码分析

@EnableCircuitBreaker分析

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@Import(EnableCircuitBreakerImportSelector.class)
public @interface EnableCircuitBreaker {

}
  • 导入了EnableCircuitBreakerImportSelector选择器
@Order(Ordered.LOWEST_PRECEDENCE - 100)
public class EnableCircuitBreakerImportSelector
		extends SpringFactoryImportSelector<EnableCircuitBreaker> {

	@Override
	protected boolean isEnabled() {
		return getEnvironment().getProperty("spring.cloud.circuit.breaker.enabled",
				Boolean.class, Boolean.TRUE);
	}

}
  • 父类SpringFactoryImportSelector
public abstract class SpringFactoryImportSelector<T>
		implements DeferredImportSelector, BeanClassLoaderAware, EnvironmentAware {

	private final Log log = LogFactory.getLog(SpringFactoryImportSelector.class);

	private ClassLoader beanClassLoader;

	private Class<T> annotationClass;

	private Environment environment;

	@SuppressWarnings("unchecked")
	protected SpringFactoryImportSelector() {
		this.annotationClass = (Class<T>) GenericTypeResolver
				.resolveTypeArgument(this.getClass(), SpringFactoryImportSelector.class);
	}

	@Override
	public String[] selectImports(AnnotationMetadata metadata) {
		if (!isEnabled()) {
			return new String[0];
		}
		AnnotationAttributes attributes = AnnotationAttributes.fromMap(
				metadata.getAnnotationAttributes(this.annotationClass.getName(), true));

		Assert.notNull(attributes, "No " + getSimpleName() + " attributes found. Is "
				+ metadata.getClassName() + " annotated with @" + getSimpleName() + "?");

		// Find all possible auto configuration classes, filtering duplicates
		List<String> factories = new ArrayList<>(new LinkedHashSet<>(SpringFactoriesLoader
				.loadFactoryNames(this.annotationClass, this.beanClassLoader)));

		if (factories.isEmpty() && !hasDefaultFactory()) {
			throw new IllegalStateException("Annotation @" + getSimpleName()
					+ " found, but there are no implementations. Did you forget to include a starter?");
		}

		if (factories.size() > 1) {
			// there should only ever be one DiscoveryClient, but there might be more than
			// one factory
			this.log.warn("More than one implementation " + "of @" + getSimpleName()
					+ " (now relying on @Conditionals to pick one): " + factories);
		}

		return factories.toArray(new String[factories.size()]);
	}

	protected boolean hasDefaultFactory() {
		return false;
	}

	protected abstract boolean isEnabled();

	protected String getSimpleName() {
		return this.annotationClass.getSimpleName();
	}

	protected Class<T> getAnnotationClass() {
		return this.annotationClass;
	}

	protected Environment getEnvironment() {
		return this.environment;
	}

	@Override
	public void setEnvironment(Environment environment) {
		this.environment = environment;
	}

	@Override
	public void setBeanClassLoader(ClassLoader classLoader) {
		this.beanClassLoader = classLoader;
	}

}

通过SpringFactoriesLoader加载了EnableCircuitBreaker的配置类

查找这个org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker配置类发现spring-cloud-netflix-hystrix包下META-INF/spring.factories有如下配置:

org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker=
org.springframework.cloud.netflix.hystrix.HystrixCircuitBreakerConfiguration

HystrixCircuitBreakerConfiguration Hystrix服务熔断配置

@Configuration(proxyBeanMethods = false)
public class HystrixCircuitBreakerConfiguration {

    // 从这个地方查看Hystrix也是通过Spring Aop进行实现的
	@Bean
	public HystrixCommandAspect hystrixCommandAspect() {
		return new HystrixCommandAspect();
	}

	@Bean
	public HystrixShutdownHook hystrixShutdownHook() {
		return new HystrixShutdownHook();
	}

	@Bean
	public HasFeatures hystrixFeature() {
		return HasFeatures
				.namedFeatures(new NamedFeature("Hystrix", HystrixCommandAspect.class));
	}

	/**
	 * {@link DisposableBean} that makes sure that Hystrix internal state is cleared when
	 * {@link ApplicationContext} shuts down.
	 */
	private class HystrixShutdownHook implements DisposableBean {

		@Override
		public void destroy() throws Exception {
			// Just call Hystrix to reset thread pool etc.
			Hystrix.reset();
		}

	}

}

HystrixCommandAspect HystrixCommand切面

@Aspect
public class HystrixCommandAspect {

    private static final Map<HystrixPointcutType, MetaHolderFactory> META_HOLDER_FACTORY_MAP;

    static {
        META_HOLDER_FACTORY_MAP = ImmutableMap.<HystrixPointcutType, MetaHolderFactory>builder()
                .put(HystrixPointcutType.COMMAND, new CommandMetaHolderFactory())
                .put(HystrixPointcutType.COLLAPSER, new CollapserMetaHolderFactory())
                .build();
    }

    @Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand)")

    public void hystrixCommandAnnotationPointcut() {
    }

    @Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCollapser)")
    public void hystrixCollapserAnnotationPointcut() {
    }

    @Around("hystrixCommandAnnotationPointcut() || hystrixCollapserAnnotationPointcut()")
    public Object methodsAnnotatedWithHystrixCommand(final ProceedingJoinPoint joinPoint) throws Throwable {
        Method method = getMethodFromTarget(joinPoint);
        Validate.notNull(method, "failed to get method from joinPoint: %s", joinPoint);
        if (method.isAnnotationPresent(HystrixCommand.class) && method.isAnnotationPresent(HystrixCollapser.class)) {
            throw new IllegalStateException("method cannot be annotated with HystrixCommand and HystrixCollapser " +
                    "annotations at the same time");
        }
        MetaHolderFactory metaHolderFactory = META_HOLDER_FACTORY_MAP.get(HystrixPointcutType.of(method));
        MetaHolder metaHolder = metaHolderFactory.create(joinPoint);
        HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder);
        ExecutionType executionType = metaHolder.isCollapserAnnotationPresent() ?
                metaHolder.getCollapserExecutionType() : metaHolder.getExecutionType();

        Object result;
        try {
            if (!metaHolder.isObservable()) {
                result = CommandExecutor.execute(invokable, executionType, metaHolder);
            } else {
                result = executeObservable(invokable, executionType, metaHolder);
            }
        } catch (HystrixBadRequestException e) {
            throw e.getCause();
        } catch (HystrixRuntimeException e) {
            throw hystrixRuntimeExceptionToThrowable(metaHolder, e);
        }
        return result;
    }
    ....
}

从代码来看切面拦截了两个注解,@HystrixCommand和@HystrixCollapser,前面已经熟悉过@HystrixCommand,这个@HystrixCollapser又是什么内容

@HystrixCollapser注解

/**
 * This annotation is used to collapse some commands into a single backend dependency call.
 * This annotation should be used together with {@link HystrixCommand} annotation.
 * <p/>
 * Example:
 * <pre>
 *     @HystrixCollapser(batchMethod = "getUserByIds"){
 *          public Future<User> getUserById(String id) {
 *          return null;
 * }
 *  @HystrixCommand
 *      public List<User> getUserByIds(List<String> ids) {
 *          List<User> users = new ArrayList<User>();
 *          for (String id : ids) {
 *              users.add(new User(id, "name: " + id));
 *          }
 *      return users;
 * }
 *   </pre>
 *
 * A method annotated with {@link HystrixCollapser} annotation can return any
 * value with compatible type, it does not affect the result of collapser execution,
 * collapser method can even return {@code null} or another stub.
 * Pay attention that if a collapser method returns parametrized Future then generic type must be equal to generic type of List,
 * for instance:
 * <pre>
 *     Future<User> - return type of collapser method
 *     List<User> - return type of batch command method
 * </pre>
 * <p/>
 * Note: batch command method must be annotated with {@link HystrixCommand} annotation.
 */
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface HystrixCollapser {

    /**
     * Specifies a collapser key.
     * <p/>
     * default => the name of annotated method.
     *
     * @return collapser key.
     */
    String collapserKey() default "";

    /**
     * Method name of batch command.
     * <p/>
     * Method must have the following signature:
     * <pre>
     *     java.util.List method(java.util.List)
     * </pre>
     * NOTE: batch method can have only one argument.
     *
     * @return method name of batch command
     */
    String batchMethod();

    /**
     * Defines what scope the collapsing should occur within.
     * <p/>
     * default => the {@link Scope#REQUEST}.
     *
     * @return {@link Scope}
     */
    Scope scope() default Scope.REQUEST;

    /**
     * Specifies collapser properties.
     *
     * @return collapser properties
     */
    HystrixProperty[] collapserProperties() default {};

}

从@HystrixCollapser这个代码注释来看是针对Future返回值来做的,理解来看是多个@HystrixCommand组合注解

Spring Cloud 官方文档没有对@HystrixCollapser注解类的介绍

/**
 * 注解演示服务端点
 *
 * @author FelixFly <[email protected]>
 * @date 2020/2/5
 */
@RestController
@RequestMapping("/annotation")
public class AnnotationDemoController {


    @HystrixCollapser(batchMethod = "batchMethod", collapserProperties = {
            @HystrixProperty(name = "timerDelayInMilliseconds", value = "50")
    })
    @GetMapping("/collapser/{message}")
    public Future<String> collapser(@PathVariable String message) {
        await();
        return null;
    }


    @HystrixCommand
    public List<String> batchMethod(List<String> messages) {
        await();
        return messages.stream().map(message ->"Thread:" + message).collect(toList());
    }

    public List<String> batchFallBack(List<String> messages) {
        return messages.stream().map(message ->"Fallback:" + message).collect(toList());
    }

    @HystrixCommand(fallbackMethod = "fallBack", commandProperties = {
            @HystrixProperty(name = "execution.isolation.strategy", value = "THREAD"),
            @HystrixProperty(name = "execution.isolation.thread.timeoutInMilliseconds", value = "50")
    })
    @GetMapping("/thread/{message}")
    public String thread(@PathVariable String message) {
        await();
        return "Thread:" + message;
    }

    @HystrixCommand(fallbackMethod = "fallBack", commandProperties = {
            @HystrixProperty(name = "execution.isolation.strategy", value = "SEMAPHORE"),
            @HystrixProperty(name = "execution.isolation.thread.timeoutInMilliseconds", value = "50"),
            @HystrixProperty(name = "execution.isolation.semaphore.maxConcurrentRequests", value = "1")
    })
    @GetMapping("/semaphore/{message}")
    public String hello(@PathVariable String message) {
        await();
        return "Semaphore:" + message;
    }

    public String fallBack(String message) {
        return "Fallback:" + message;
    }

    public void await() {
        int waitTime = new Random().nextInt(100);
        try {
            System.out.printf("线程【%s】执行时间%d毫秒\n", Thread.currentThread().getName(), waitTime);
            TimeUnit.MILLISECONDS.sleep(waitTime);
        } catch (InterruptedException e) {
            throw new RuntimeException("线程中断");
        }
    }

}

@HystrixCollapser配置信息:https://github.com/Netflix/Hystrix/wiki/Configuration#Collapser

访问地址:http://127.0.0.1:8090/annotation/collapser/123

{
	"cancelled": false,
	"done": true
}

暂时还没发现具体的用处,等发现的时候来继续补充

原文:https://gper.club/articles/7e7e7f7ff4g5egc7g6f