Hystrix的使用

@SpringBootApplication
@EnableEurekaClient
//开启Hystrix断路器功能
@EnableHystrix
public class OrderApplication {

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

@GetMapping("/query")
public String query();

@HystrixCommand
public String query() {
    System.out.println("调用查询接口，线程号：" + Thread.currentThread().getId());
    return "查询接口";
}

@RunWith(SpringRunner.class)
@SpringBootTest(classes = {OrderApplication.class})
public class HystrixTest {
    //线程池 默认100个线程
    private ExecutorService executorService = Executors.newFixedThreadPool(100);
    //  并发量通知执行多少个请求
    private int concurrentNumber = 15;
    //使用CountDownLatch进行并发请求
    private CountDownLatch countDownLatch = new CountDownLatch(concurrentNumber);


    @Autowired
    private OrderService orderService;

    @Test
    public void test() {
        for (int i = 0; i < concurrentNumber; i++) {
            executorService.execute(() -> {
                countDownLatch.countDown();
                String value = orderService.query();
                System.out.println("开始查询接口，调用线程号：" + Thread.currentThread().getId() + ",返回结果：" + value);
            });
        }
        //结束并等待所以子线程结束任务否则会到这里直接退出不会等待子线程打印结果
        executorService.shutdown();
        while (!executorService.isTerminated()) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

@HystrixCommand(fallbackMethod = "fallback", commandKey = "queryKeys", groupKey = "groupKey")
public String query() {
    System.out.println("调用查询接口，线程号：" + Thread.currentThread().getId());
    return "调用查询接口...";
}

/**
  * 快速失败方法
  *
  * @return
  */
public String fallback() {
  	System.out.println("调用快速失败方法，线程号："+Thread.currentThread().getId());
    return "调用快速失败方法...";
}

@HystrixCommand(fallbackMethod = "fallback", commandKey = "queryKeys", groupKey = "groupKey",
                commandProperties = {
                    //线程池隔离策略 默认值 可以省略
                    @HystrixProperty(name = "execution.isolation.strategy", value = "THREAD")
                },
                threadPoolKey = "queryOrderHystrixPool",
                threadPoolProperties = {
                    //核心线程数设置为10 默认值
                    @HystrixProperty(name = "coreSize", value = "10")
                })
public String query() {
    System.out.println("调用查询接口，线程号：" + Thread.currentThread().getId());
    return "调用查询接口...";
}

@HystrixCommand(fallbackMethod = "fallback", commandKey = "queryKeys", groupKey = "groupKey",
                commandProperties = {
                    //线程池隔离策略 默认值 可以省略
                    @HystrixProperty(name = "execution.isolation.strategy", value = "SEMAPHORE"),
                    //这参数是用来控制信号量隔离级别的并发大小的
                    @HystrixProperty(name = "execution.isolation.semaphore.maxConcurrentRequests", value = "10")
                })
public String query() {
    System.out.println("调用查询接口，线程号：" + Thread.currentThread().getId());
    return "调用查询接口...";
}

/**
  * 快速失败方法
  *
  * @return
  */
public String fallback() {
    System.out.println("调用快速失败方法，线程号："+Thread.currentThread().getId());
    return "调用快速失败方法...";
}

@HystrixCommand(fallbackMethod = "fallback", commandKey = "queryKeys", groupKey = "groupKey",
                commandProperties = {
                    //线程池隔离策略 默认值 可以省略
                    @HystrixProperty(name = "execution.isolation.strategy", value = "SEMAPHORE"),
                    //这参数是用来控制信号量隔离级别的并发大小的
                    @HystrixProperty(name = "execution.isolation.semaphore.maxConcurrentRequests", value = "10")
                })
public String query() {
    System.out.println("调用查询接口，线程号：" + Thread.currentThread().getId());
    if (Thread.currentThread().getId() % 2 == 0) {
        System.out.println("出现了异常，线程号：" + Thread.currentThread().getId());
        int k = 1 / 0;
    }
    return "调用查询接口...";
}

/**
  * 降级方法
  *
  * @return
  */
public String fallback() {
    System.out.println("调用降级方法，线程号：" + Thread.currentThread().getId());
    return "调用降级方法...";
}

@SpringBootApplication
//开启Hystrix仪表盘监控注解 监控地址 http://localhost:9999/hystrix
@EnableHystrixDashboard
public class HystrixDashboardApplication {
    public static void main(String[] args) {
        SpringApplication.run(HystrixDashboardApplication.class);
    }
}

public final class DelegatingUserContextCallable<V> implements Callable<V> {
    private final Callable<V> delegate;
    // 用户信息上下文（根据项目实际情况定义ThreadLocal上下文）
    private UserContext originalUserContext;

    public DelegatingUserContextCallable(Callable<V> delegate,
                                         UserContext userContext) {
        this.delegate = delegate;
        this.originalUserContext = userContext;
    }

    public V call() throws Exception {
        // 将当前的用户上下文设置进Hystrix线程的TreadLocal中
        UserContextHolder.setContext(originalUserContext);
        try {
            return delegate.call();
        }
        finally {
            // 执行完毕，记得清理ThreadLocal资源
            this.originalUserContext = null;
        }
    }

    public static <V> Callable<V> create(Callable<V> delegate,
                                         UserContext userContext) {
        return new DelegatingUserContextCallable<V>(delegate, userContext);
    }
}

public class ThreadLocalAwareStrategy extends HystrixConcurrencyStrategy {
    
    // 最简单的方式就是引入现有的并发策略，进行功能扩展
    private final HystrixConcurrencyStrategy existingConcurrencyStrategy;

    public ThreadLocalAwareStrategy(
            HystrixConcurrencyStrategy existingConcurrencyStrategy) {
        this.existingConcurrencyStrategy = existingConcurrencyStrategy;
    }

    @Override
    public BlockingQueue<Runnable> getBlockingQueue(int maxQueueSize) {
        return existingConcurrencyStrategy != null
                ? existingConcurrencyStrategy.getBlockingQueue(maxQueueSize)
                : super.getBlockingQueue(maxQueueSize);
    }

    @Override
    public <T> HystrixRequestVariable<T> getRequestVariable(
            HystrixRequestVariableLifecycle<T> rv) {
        return existingConcurrencyStrategy != null
                ? existingConcurrencyStrategy.getRequestVariable(rv)
                : super.getRequestVariable(rv);
    }

    @Override
    public ThreadPoolExecutor getThreadPool(HystrixThreadPoolKey threadPoolKey,
                                            HystrixProperty<Integer> corePoolSize,
                                            HystrixProperty<Integer> maximumPoolSize,
                                            HystrixProperty<Integer> keepAliveTime, TimeUnit unit,
                                            BlockingQueue<Runnable> workQueue) {
        return existingConcurrencyStrategy != null
                ? existingConcurrencyStrategy.getThreadPool(threadPoolKey, corePoolSize,
                maximumPoolSize, keepAliveTime, unit, workQueue)
                : super.getThreadPool(threadPoolKey, corePoolSize, maximumPoolSize,
                keepAliveTime, unit, workQueue);
    }

    @Override
    public <T> Callable<T> wrapCallable(Callable<T> callable) {
        return existingConcurrencyStrategy != null
                ? existingConcurrencyStrategy
                .wrapCallable(new DelegatingUserContextCallable<>(callable, UserContextHolder.getContext()))
                : super.wrapCallable(new DelegatingUserContextCallable<T>(callable, UserContextHolder.getContext()));
    }
}

@ConditionalOnClass(HystrixConcurrencyStrategy.class)
@Configuration
public class HystrixThreadLocalConfiguration {

    @Autowired(required = false)
    private HystrixConcurrencyStrategy existingConcurrencyStrategy;

    @PostConstruct
    public void init() {
        // Keeps references of existing Hystrix plugins.
        HystrixEventNotifier eventNotifier = HystrixPlugins.getInstance()
                .getEventNotifier();
        HystrixMetricsPublisher metricsPublisher = HystrixPlugins.getInstance()
                .getMetricsPublisher();
        HystrixPropertiesStrategy propertiesStrategy = HystrixPlugins.getInstance()
                .getPropertiesStrategy();
        HystrixCommandExecutionHook commandExecutionHook = HystrixPlugins.getInstance()
                .getCommandExecutionHook();

        HystrixPlugins.reset();

        HystrixPlugins.getInstance().registerConcurrencyStrategy(new ThreadLocalAwareStrategy(existingConcurrencyStrategy));
        HystrixPlugins.getInstance().registerEventNotifier(eventNotifier);
        HystrixPlugins.getInstance().registerMetricsPublisher(metricsPublisher);
        HystrixPlugins.getInstance().registerPropertiesStrategy(propertiesStrategy);
        HystrixPlugins.getInstance().registerCommandExecutionHook(commandExecutionHook);
    }
}