Hystrix源码解析

良辰美景TT

发布于 2018-09-11 14:30:52

2K0

发布于 2018-09-11 14:30:52

文章被收录于专栏：java、Spring、技术分享java、Spring、技术分享

Hystrix简介

在分布式系统中，难免有对外部接口的依赖，而外部接口有可能出现响应缓慢，大量请求超时，大量访问出现异常等情况。出现上面所说的情况有可能是由很多原因导制的，可能是网络抖动，外部系统有没有测出的bug，系统遭遇黑客攻击等。因为一个接口的异常，有可能导制线程阻塞，影响到其它接口的服务，甚至整个系统的服务给拖跨，对外部系统依赖的模块越多，出现的风险也就会越高，Hystrix正是用于解决这样的问题。Hystrix同样是Netflix公司开源的用于解决分布式问题而开源的框架。源码网址为：https://github.com/Netflix/Hystrix。Hystrix提供了如下几种解决方案应对上面说的问题，分别为：

线程池隔离
信号量隔离
熔断
降级回退

Hystrix 版的 Hello World

在pom.xml文件里引入Hystrix依赖的类

   <dependencies>
       <dependency>
           <groupId>com.netflix.hystrix</groupId>
           <artifactId>hystrix-core</artifactId>
           <version>1.5.13</version>
       </dependency>
   </dependencies>

编写业务Command

package com.ivan.client.hystrix;

import com.netflix.hystrix.HystrixCommand;
import com.netflix.hystrix.HystrixCommandGroupKey;

public class HelloCommand extends HystrixCommand<String> {

   protected HelloCommand() {
       super(HystrixCommandGroupKey.Factory.asKey("test"));
   }

   @Override
   protected String run() throws Exception {
       //模拟请求外部接口需要的时间长度
       Thread.sleep(500);
       return "sucess";
   }
   
   @Override
   protected String getFallback() {
     //当外部请求超时后，会执行fallback里的业务逻辑
     System.out.println("执行了回退方法");
     return "error";
   }

}

模拟系统调用

package com.ivan.client.hystrix;

public class App {
    public static void main(String[] args) {
        HelloCommand command = new HelloCommand();
        String result = command.execute();
        System.out.println(result);
    }
}

当我们增大 HelloCommand run方法里Thread.sleep()方法的时长时，我们可以看到 command.execute()方法调用返回了error。在实际的使用中，当发现第三方接口调用不通的情况下，会调用fallback方法进行降级处理，比如可以返回一段错误提示。

Hystrix线程池隔离

在分布式的系统里，系统可能对多个外部系统都有依赖关系，比同订单系统同时对会员系统，库存系统统，优惠券系统都有依赖。假如优惠券系统出现访问异常的时候，会超成线程的堆积，对于系统调用库存系统与会员系统的业务也不可用。而通过线程池能够将不同的业务由不同的线程池处理，从而做到保护其它业务能够正常访问。下面就来看看Hystrix是根据什么来创建线程的。

找到HystrixCommand的父类AbstractCommand，里面有个构造方法，从构造方法可以看出里这里定义了 threadPool对象。代码如下，关键代码都有做相应的注释

/**
这个方法是AbstractCommand的构造方法，里面用于初使化AbstractCommand，包括circuitBreaker 与线程池对象都在这里进行构造
**/
    protected AbstractCommand(HystrixCommandGroupKey group, HystrixCommandKey key, HystrixThreadPoolKey threadPoolKey, HystrixCircuitBreaker circuitBreaker, HystrixThreadPool threadPool,
            HystrixCommandProperties.Setter commandPropertiesDefaults, HystrixThreadPoolProperties.Setter threadPoolPropertiesDefaults,
            HystrixCommandMetrics metrics, TryableSemaphore fallbackSemaphore, TryableSemaphore executionSemaphore,
            HystrixPropertiesStrategy propertiesStrategy, HystrixCommandExecutionHook executionHook) {
//commandGroup对象，用于组织一类业务相关的对象
        this.commandGroup = initGroupKey(group);
// commandKey默认是以类为为名称的
        this.commandKey = initCommandKey(key, getClass());
        this.properties = initCommandProperties(this.commandKey, propertiesStrategy, commandPropertiesDefaults);
//这个方法里定义了TheradPool里的关键字，默认以传入的commandGroup 的name做为key的名称
        this.threadPoolKey = initThreadPoolKey(threadPoolKey, this.commandGroup, this.properties.executionIsolationThreadPoolKeyOverride().get());
        this.metrics = initMetrics(metrics, this.commandGroup, this.threadPoolKey, this.commandKey, this.properties);
        this.circuitBreaker = initCircuitBreaker(this.properties.circuitBreakerEnabled().get(), circuitBreaker, this.commandGroup, this.commandKey, this.properties, this.metrics);
//这里就是线程池对象啦。
        this.threadPool = initThreadPool(threadPool, this.threadPoolKey, threadPoolPropertiesDefaults);

        //Strategies from plugins
        this.eventNotifier = HystrixPlugins.getInstance().getEventNotifier();
        this.concurrencyStrategy = HystrixPlugins.getInstance().getConcurrencyStrategy();
        HystrixMetricsPublisherFactory.createOrRetrievePublisherForCommand(this.commandKey, this.commandGroup, this.metrics, this.circuitBreaker, this.properties);
        this.executionHook = initExecutionHook(executionHook);

        this.requestCache = HystrixRequestCache.getInstance(this.commandKey, this.concurrencyStrategy);
        this.currentRequestLog = initRequestLog(this.properties.requestLogEnabled().get(), this.concurrencyStrategy);

        /* fallback semaphore override if applicable */
        this.fallbackSemaphoreOverride = fallbackSemaphore;

        /* execution semaphore override if applicable */
        this.executionSemaphoreOverride = executionSemaphore;
    }

/**
这个方法用于得到HystrixThreadPoolKey 对象， Hystrix内部有大量的Key对象，可以简单理解这些  Key都是相应对象的唯一标识。从代码里可以看出，默认情况下Hystrix采用的是commandGroup 的name做为Thread Pool的key值。
**/
    private static HystrixThreadPoolKey initThreadPoolKey(HystrixThreadPoolKey threadPoolKey, HystrixCommandGroupKey groupKey, String threadPoolKeyOverride) {
        if (threadPoolKeyOverride == null) {
            // we don't have a property overriding the value so use either HystrixThreadPoolKey or HystrixCommandGroup
            if (threadPoolKey == null) {
                /* use HystrixCommandGroup if HystrixThreadPoolKey is null */
                return HystrixThreadPoolKey.Factory.asKey(groupKey.name());
            } else {
                return threadPoolKey;
            }
        } else {
            // we have a property defining the thread-pool so use it instead
            return HystrixThreadPoolKey.Factory.asKey(threadPoolKeyOverride);
        }
    }

/**
在这里将调用具体的构造线程池的方法。
**/
    private static HystrixThreadPool initThreadPool(HystrixThreadPool fromConstructor, HystrixThreadPoolKey threadPoolKey, HystrixThreadPoolProperties.Setter threadPoolPropertiesDefaults) {
        if (fromConstructor == null) {
            // get the default implementation of HystrixThreadPool
            return HystrixThreadPool.Factory.getInstance(threadPoolKey, threadPoolPropertiesDefaults);
        } else {
            return fromConstructor;
        }
    }

从上面的代码分析我们知道线程池的构造最终会落到HystrixThreadPool.Factory这个类上面。这个类内存持有一个ConcurrentHashMap用于缓存线程池对象，当传入的HystrixThreadPoolKey已经构造过了相应的ThreadPool，将会直接从ConcurrentHashMap里返回已经生成的ThreadPool。如果传入的HystrixThreadPoolKey没有相应的ThreadPool，将构造新的ThreadPool并放入到ConcurrentHashMap这个缓存对象上。下面是关键代码：

static class Factory {
  final static ConcurrentHashMap<String, HystrixThreadPool> threadPools = new ConcurrentHashMap<String, HystrixThreadPool>();

static HystrixThreadPool getInstance(HystrixThreadPoolKey threadPoolKey, HystrixThreadPoolProperties.Setter propertiesBuilder) {
            // get the key to use instead of using the object itself so that if people forget to implement equals/hashcode things will still work
            String key = threadPoolKey.name();

            // this should find it for all but the first time
//这里从缓存取
            HystrixThreadPool previouslyCached = threadPools.get(key);
            if (previouslyCached != null) {
                return previouslyCached;
            }

            // if we get here this is the first time so we need to initialize
//这里需要保证线程安全，加上了相应的锁
            synchronized (HystrixThreadPool.class) {
                if (!threadPools.containsKey(key)) {
//具体的线程池是由HystrixThreadPoolDefault进行构造的
                    threadPools.put(key, new HystrixThreadPoolDefault(threadPoolKey, propertiesBuilder));
                }
            }
            return threadPools.get(key);
        }
}

HystrixThreadPoolDefault 内部通过HystrixConcurrencyStrategy这个对象进行线程池的构造，里面根据传入的properties信息来构造线程池对象。关键代码如下：

static class HystrixThreadPoolDefault implements HystrixThreadPool {
        private static final Logger logger = LoggerFactory.getLogger(HystrixThreadPoolDefault.class);

        private final HystrixThreadPoolProperties properties;
        private final BlockingQueue<Runnable> queue;
        private final ThreadPoolExecutor threadPool;
        private final HystrixThreadPoolMetrics metrics;
        private final int queueSize;

        public HystrixThreadPoolDefault(HystrixThreadPoolKey threadPoolKey, HystrixThreadPoolProperties.Setter propertiesDefaults) {
            this.properties = HystrixPropertiesFactory.getThreadPoolProperties(threadPoolKey, propertiesDefaults);
            HystrixConcurrencyStrategy concurrencyStrategy = HystrixPlugins.getInstance().getConcurrencyStrategy();
            this.queueSize = properties.maxQueueSize().get();

            this.metrics = HystrixThreadPoolMetrics.getInstance(threadPoolKey,
                    concurrencyStrategy.getThreadPool(threadPoolKey, properties),
                    properties);
            this.threadPool = this.metrics.getThreadPool();
            this.queue = this.threadPool.getQueue();

            /* strategy: HystrixMetricsPublisherThreadPool */
            HystrixMetricsPublisherFactory.createOrRetrievePublisherForThreadPool(threadPoolKey, this.metrics, this.properties);
        }
}

HystrixConcurrencyStrategy 类里我们可以看到采用的我们熟悉的ThreadPoolExecutor对象来构造线程池。里面需要传入核心线程池的大小，最大线程数，队列等关键信息。关键代码如下：

    public ThreadPoolExecutor getThreadPool(final HystrixThreadPoolKey threadPoolKey, HystrixThreadPoolProperties threadPoolProperties) {
        final ThreadFactory threadFactory = getThreadFactory(threadPoolKey);

        final boolean allowMaximumSizeToDivergeFromCoreSize = threadPoolProperties.getAllowMaximumSizeToDivergeFromCoreSize().get();
        final int dynamicCoreSize = threadPoolProperties.coreSize().get();
        final int keepAliveTime = threadPoolProperties.keepAliveTimeMinutes().get();
        final int maxQueueSize = threadPoolProperties.maxQueueSize().get();
        final BlockingQueue<Runnable> workQueue = getBlockingQueue(maxQueueSize);

        if (allowMaximumSizeToDivergeFromCoreSize) {
            final int dynamicMaximumSize = threadPoolProperties.maximumSize().get();
            if (dynamicCoreSize > dynamicMaximumSize) {
                logger.error("Hystrix ThreadPool configuration at startup for : " + threadPoolKey.name() + " is trying to set coreSize = " +
                        dynamicCoreSize + " and maximumSize = " + dynamicMaximumSize + ".  Maximum size will be set to " +
                        dynamicCoreSize + ", the coreSize value, since it must be equal to or greater than the coreSize value");
                return new ThreadPoolExecutor(dynamicCoreSize, dynamicCoreSize, keepAliveTime, TimeUnit.MINUTES, workQueue, threadFactory);
            } else {
                return new ThreadPoolExecutor(dynamicCoreSize, dynamicMaximumSize, keepAliveTime, TimeUnit.MINUTES, workQueue, threadFactory);
            }
        } else {
            return new ThreadPoolExecutor(dynamicCoreSize, dynamicCoreSize, keepAliveTime, TimeUnit.MINUTES, workQueue, threadFactory);
        }
    }

从上面代码的分析我们可以得出，线程池是以HystrixCommandGroupKey进行划分的，不同的CommandGroup有不同的线程池来处理，而这个CommandGroup在我们的分布式系统中，可以把相关的业务处理放到一个CommandGroup中。

Hystrix熔断

熔断器，现实生活中有一个很好的类比，就是家庭电路中都会安装一个保险盒，当电流过大的时候保险盒里面的保险丝会自动断掉，来保护家里的各种电器及电路。Hystrix中的熔断器(Circuit Breaker)也是起到这样的作用，Hystrix在运行过程中会向每个CommandKey对应的熔断器报告成功、失败、超时和拒绝的状态，熔断器维护计算统计的数据，根据这些统计的信息来确定熔断器是否打开。如果打开，后续的请求都会被截断（不再执行run方法里的内容了，直接执行fallback方法里的内容）。然后会隔一段时间默认是5s，尝试半开，放入一部分流量请求进来，相当于对依赖服务进行一次健康检查，如果恢复，熔断器关闭，随后完全恢复调用。改造HelloCommand类代码如下：

package com.ivan.client.hystrix;

import com.netflix.hystrix.HystrixCommand;
import com.netflix.hystrix.HystrixCommandGroupKey;
import com.netflix.hystrix.HystrixCommandProperties;

public class HelloCommand extends HystrixCommand<String> {

    protected HelloCommand() {
        
        
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("test"))
                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
                        //开启熔断模式
                        .withCircuitBreakerEnabled(true)
                        //出现错误的比率超过30%就开启熔断
                        .withCircuitBreakerErrorThresholdPercentage(30)
                        //至少有10个请求才进行errorThresholdPercentage错误百分比计算
                        .withCircuitBreakerRequestVolumeThreshold(10)
                        //半开试探休眠时间，这里设置为3秒
                        .withCircuitBreakerSleepWindowInMilliseconds(3000)
                        )
                );
        
    }

    @Override
    protected String run() throws Exception {
        //模拟外部请求需要的时间长度
        System.out.println("执行了run方法");
        Thread.sleep(2000);
        return "sucess";
    }
    
    @Override
    protected String getFallback() {
      //当外部请求超时后，会执行fallback里的业务逻辑
      System.out.println("执行了回退方法");
      return "error";
    }

}

改造App的代码，通过执行30次请求，可以看出刚开始的时候（前10次请求）会执行run方法里的逻辑，一量熔断器打开后，将不执行run方法里的内容，而是直接执行getFallback方法里的逻辑，直到过了设置的3秒后才又有流量执行run方法的逻辑。改造App后的代码如下：

package com.ivan.client.hystrix;

public class App {
    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 30; i++) {
            HelloCommand command = new HelloCommand();
            String result = command.execute();
            System.out.println("circuit Breaker is open : " + command.isCircuitBreakerOpen());
            if(command.isCircuitBreakerOpen()){
                Thread.currentThread().sleep(500);
            }
        }
    }
}

Hystrix熔断源码分析

找到AbstractCommand类的initCircuitBreaker方法，这是熔断器的构造方法入口。首先判断是否打开了熔断器，只有在打开了熔断器后才会通过HystrixCircuitBreaker.Factory工厂新建一个熔断器，源码如下：

    private static HystrixCircuitBreaker initCircuitBreaker(boolean enabled, HystrixCircuitBreaker fromConstructor,
                                                            HystrixCommandGroupKey groupKey, HystrixCommandKey commandKey,
                                                            HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
        if (enabled) {
            if (fromConstructor == null) {
                // get the default implementation of HystrixCircuitBreaker
                return HystrixCircuitBreaker.Factory.getInstance(commandKey, groupKey, properties, metrics);
            } else {
                return fromConstructor;
            }
        } else {
            return new NoOpCircuitBreaker();
        }
    }

HystrixCircuitBreaker.Factory 类里对熔断器根据CommandKey进行了缓存，如果存在直接取缓存里的key,不存在则新建HystrixCircuitBreakerImpl对象用于熔断操作。源代码如下：

    class Factory {
      //circuitBreakersByCommand 是个ConcurrentHashMap， 这里缓存了系统的所有熔断器
        private static ConcurrentHashMap<String, HystrixCircuitBreaker> circuitBreakersByCommand = new ConcurrentHashMap<String, HystrixCircuitBreaker>();

        public static HystrixCircuitBreaker getInstance(HystrixCommandKey key, HystrixCommandGroupKey group, HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
            // this should find it for all but the first time
  //先从缓存里取
            HystrixCircuitBreaker previouslyCached = circuitBreakersByCommand.get(key.name());
            if (previouslyCached != null) {
                return previouslyCached;
            }

            // if we get here this is the first time so we need to initialize

            // Create and add to the map ... use putIfAbsent to atomically handle the possible race-condition of
            // 2 threads hitting this point at the same time and let ConcurrentHashMap provide us our thread-safety
            // If 2 threads hit here only one will get added and the other will get a non-null response instead.
 //取不到对象才会创建个HystrixCircuitBreakerImpl对象并放入缓存Map中
            HystrixCircuitBreaker cbForCommand = circuitBreakersByCommand.putIfAbsent(key.name(), new HystrixCircuitBreakerImpl(key, group, properties, metrics));
            if (cbForCommand == null) {
                // this means the putIfAbsent step just created a new one so let's retrieve and return it
                return circuitBreakersByCommand.get(key.name());
            } else {
                // this means a race occurred and while attempting to 'put' another one got there before
                // and we instead retrieved it and will now return it
                return cbForCommand;
            }
        }

        /**
         * Get the {@link HystrixCircuitBreaker} instance for a given {@link HystrixCommandKey} or null if none exists.
         * 
         * @param key
         *            {@link HystrixCommandKey} of {@link HystrixCommand} instance requesting the {@link HystrixCircuitBreaker}
         * @return {@link HystrixCircuitBreaker} for {@link HystrixCommandKey}
         */
        public static HystrixCircuitBreaker getInstance(HystrixCommandKey key) {
            return circuitBreakersByCommand.get(key.name());
        }

        /**
         * Clears all circuit breakers. If new requests come in instances will be recreated.
         */
        /* package */static void reset() {
            circuitBreakersByCommand.clear();
        }
    }

HystrixCircuitBreakerImpl 这个类里定义了一个状态变量，断路由有三种状态，分别为关闭，打开，半开状态。重点关注下allowRequest方法，在allowRequest里首先判断forceOpen属性是否打开，如果打开则不允许有请求进入，然后forceClosed属性，如果这个属性为true,刚对所有的求求放行，相当于熔断器不起作用。之后就是状态判断了。isAfterSleepWindow（）方法用于放行超过了指定时间后的流量，。具体代码如下，关键部分有相应的注释：

class HystrixCircuitBreakerImpl implements HystrixCircuitBreaker {
        private final HystrixCommandProperties properties;
        private final HystrixCommandMetrics metrics;
  //三种状态通过枚举来定义
        enum Status {
            CLOSED, OPEN, HALF_OPEN;
        }
//状态变时，默认是关闭的状态
        private final AtomicReference<Status> status = new AtomicReference<Status>(Status.CLOSED);
//最后一次访问的时间，用于试探请求是否恢复
        private final AtomicLong circuitOpened = new AtomicLong(-1);
        private final AtomicReference<Subscription> activeSubscription = new AtomicReference<Subscription>(null);

        protected HystrixCircuitBreakerImpl(HystrixCommandKey key, HystrixCommandGroupKey commandGroup, final HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
            this.properties = properties;
            this.metrics = metrics;

            //On a timer, this will set the circuit between OPEN/CLOSED as command executions occur
            Subscription s = subscribeToStream();
            activeSubscription.set(s);
        }

        private Subscription subscribeToStream() {
            /*
             * This stream will recalculate the OPEN/CLOSED status on every onNext from the health stream
             */
            return metrics.getHealthCountsStream()
                    .observe()
                    .subscribe(new Subscriber<HealthCounts>() {
                        @Override
                        public void onCompleted() {

                        }

                        @Override
                        public void onError(Throwable e) {

                        }

                        @Override
                        public void onNext(HealthCounts hc) {
                            // check if we are past the statisticalWindowVolumeThreshold
                            if (hc.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {
                                // we are not past the minimum volume threshold for the stat window,
                                // so no change to circuit status.
                                // if it was CLOSED, it stays CLOSED
                                // if it was half-open, we need to wait for a successful command execution
                                // if it was open, we need to wait for sleep window to elapse
                            } else {
                                if (hc.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) {
                                    //we are not past the minimum error threshold for the stat window,
                                    // so no change to circuit status.
                                    // if it was CLOSED, it stays CLOSED
                                    // if it was half-open, we need to wait for a successful command execution
                                    // if it was open, we need to wait for sleep window to elapse
                                } else {
                                    // our failure rate is too high, we need to set the state to OPEN
                                    if (status.compareAndSet(Status.CLOSED, Status.OPEN)) {
                                        circuitOpened.set(System.currentTimeMillis());
                                    }
                                }
                            }
                        }
                    });
        }
//将熔断器置于关闭状态，并重置统计数据
        @Override
        public void markSuccess() {
            if (status.compareAndSet(Status.HALF_OPEN, Status.CLOSED)) {
                //This thread wins the race to close the circuit - it resets the stream to start it over from 0
                metrics.resetStream();
                Subscription previousSubscription = activeSubscription.get();
                if (previousSubscription != null) {
                    previousSubscription.unsubscribe();
                }
                Subscription newSubscription = subscribeToStream();
                activeSubscription.set(newSubscription);
                circuitOpened.set(-1L);
            }
        }

//将熔断器置于打开状态
        @Override
        public void markNonSuccess() {
            if (status.compareAndSet(Status.HALF_OPEN, Status.OPEN)) {
                //This thread wins the race to re-open the circuit - it resets the start time for the sleep window
                circuitOpened.set(System.currentTimeMillis());
            }
        }
//用于判断熔断器是否打开
        @Override
        public boolean isOpen() {
            if (properties.circuitBreakerForceOpen().get()) {
                return true;
            }
            if (properties.circuitBreakerForceClosed().get()) {
                return false;
            }
            return circuitOpened.get() >= 0;
        }
//用于判断是否放行流量
        @Override
        public boolean allowRequest() {
            if (properties.circuitBreakerForceOpen().get()) {
                return false;
            }
            if (properties.circuitBreakerForceClosed().get()) {
                return true;
            }
//第一次请求肯定就放行了
            if (circuitOpened.get() == -1) {
                return true;
            } else {
//半开状态将不放行
                if (status.get().equals(Status.HALF_OPEN)) {
                    return false;
                } else {
                    return isAfterSleepWindow();
                }
            }
        }

//根据当前时间与最后一次请求的时候进行比较，当超过了设置的SleepWindowInMilliseconds，将放行请求用于试探服务访问是否OK
        private boolean isAfterSleepWindow() {
            final long circuitOpenTime = circuitOpened.get();
            final long currentTime = System.currentTimeMillis();
            final long sleepWindowTime = properties.circuitBreakerSleepWindowInMilliseconds().get();
            return currentTime > circuitOpenTime + sleepWindowTime;
        }

//用于试探服务是否OK的方法
        @Override
        public boolean attemptExecution() {
            if (properties.circuitBreakerForceOpen().get()) {
                return false;
            }
            if (properties.circuitBreakerForceClosed().get()) {
                return true;
            }
            if (circuitOpened.get() == -1) {
                return true;
            } else {
                if (isAfterSleepWindow()) {
                    //only the first request after sleep window should execute
                    //if the executing command succeeds, the status will transition to CLOSED
                    //if the executing command fails, the status will transition to OPEN
                    //if the executing command gets unsubscribed, the status will transition to OPEN
                    if (status.compareAndSet(Status.OPEN, Status.HALF_OPEN)) {
                        return true;
                    } else {
                        return false;
                    }
                } else {
                    return false;
                }
            }
        }
    }

Hystrix降级处理

所谓降级，就是指在在Hystrix执行非核心链路功能失败的情况下，我们如何处理，比如我们返回默认值等。如果我们要回退或者降级处理，代码上需要实现HystrixCommand.getFallback()方法或者是HystrixObservableCommand. resumeWithFallback()。

Hystrix与Spring Cloud整合

在实际项目的开发中，都会用到Fegin，所以这里的集成是在Feign的基础上进行的。

首先还是需要引入包，pom.xml文件如下：

    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>Edgware.SR4</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>

    <dependencies>
        <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-feign</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-hystrix</artifactId>
        </dependency>
    </dependencies>

启动类需要加上EnableCircuitBreaker注解代码如下：

package com.ivan.client.feign;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker;
import org.springframework.cloud.netflix.eureka.EnableEurekaClient;
import org.springframework.cloud.netflix.feign.EnableFeignClients;

@SpringBootApplication
@EnableEurekaClient
@EnableFeignClients
@EnableCircuitBreaker
public class App {

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

FeignClient 注解需有fallback属性，属性的值是个class,这个 class在是断路器打开后，会执行的业务逻辑，一般在项目里返回一个默认值。这个类需要实现与FeignClient注释上相同的接口

package com.ivan.client.feign.service;

import org.springframework.cloud.netflix.feign.FeignClient;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;

import com.ivan.client.feign.entity.User;
import com.ivan.client.feign.hystrix.fallback.UserServiceFallback;


@FeignClient(value="provider", fallback=UserServiceFallback.class)
public interface UserService {

    @RequestMapping(method = RequestMethod.GET, value = "/user/{id}")
    public User getUser(@PathVariable("id") Integer id);

    @RequestMapping(method = RequestMethod.POST, value = "/user/create", consumes = MediaType.APPLICATION_JSON_VALUE)
    public User create(User user);

}

fallback类的代码如下：

package com.ivan.client.feign.hystrix.fallback;

import org.springframework.stereotype.Component;

import com.ivan.client.feign.entity.User;
import com.ivan.client.feign.service.UserService;

@Component
public class UserServiceFallback implements UserService {

    public User getUser(Integer id) {
        System.out.println(Thread.currentThread().getName());
        System.out.println("=====执行到了fallback方法=======");
        User user = new User();
        user.setId(0);
        return user;
    }

    public User create(User user) {
        // TODO Auto-generated method stub
        return null;
    }

}

application.properties文件里需要把feign.hystrix.enabled=true 这个属性打开。配置文件如下：

server.port=9000
spring.application.name=consumer-feign-hystrix
eureka.instance.hostname=localhost
eureka.client.serviceUrl.defaultZone=http://localhost:8761/eureka/
spring.cloud.circuit.breaker.enabled=true

ribbon.ReadTimeout=5000

feign.hystrix.enabled=true
#command相关
hystrix.command.default.execution.isolation.strategy=THREAD
#设置调用者的超时时间
hystrix.command.default.execution.isolation.thread.timeoutInMilliseconds=6000
#是否开启超时设置
hystrix.command.default.execution.timeout.enabled=true
#表示是否在执行超时时，中断HystrixCommand.run() 的执行
hystrix.command.default.execution.isolation.thread.interruptOnTimeout=true

#fallback相关
#是否开启fallback功能
hystrix.command.default.fallback.enabled=true

#断路器相关
#是否开启断路器
hystrix.command.default.circuitBreaker.enabled=true
#窗口时间内打开断路器最小的请求量
hystrix.command.default.circuitBreaker.requestVolumeThreshold=5
#断路器跳闸后，在此值的时间的内，hystrix会拒绝新的请求，只有过了这个时间断路器才会打开闸门
hystrix.command.default.circuitBreaker.sleepWindowInMilliseconds=5
#失败百分比的阈值
hystrix.command.default.circuitBreaker.errorThresholdPercentage=20

#线程相关配置
#核心线程数
hystrix.threadpool.default.coreSize=5
#最大线程数
hystrix.threadpool.default.maximumSize=5
#队列的大小
hystrix.threadpool.default.maxQueueSize=1024
#因为maxQueueSize值不能被动态修改，所有通过设置此值可以实现动态修改等待队列长度。即等待的队列的数量大于queueSizeRejectionThreshold时（但是没有达到maxQueueSize值），则开始拒绝后续的请求进入队列
hystrix.threadpool.default.queueSizeRejectionThreshold=128
#设置线程多久没有服务后，需要释放（maximumSize-coreSize ）个线程
hystrix.threadpool.default.keepAliveTimeMinutes=60

上面的属性基本上包括了大部分会在项目中使用的属性，有以下几点需要重点关注一下：

上面属性的default可以改成ComandKey，这样就可以对特定的接口进行配置了，Feign中CommandKey的值为：接口名#方法名(参数类型)，如上的CommandKey为UserService#getUser(Integer)
在测试hystrix.command.default.execution.isolation.thread.timeoutInMilliseconds 属性的时候，服务端如果在指定的时间返回了结果，但系统还是调用了fallback里的逻辑，需要指定ribbon.ReadTimeout的时间。

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2018.07.10 ，如有侵权请联系 cloudcommunity@tencent.com 删除

分布式