@Service
public class HelloService {
    @Autowired
    private RestTemplate restTemplate;
    //该注解对该方法创建了熔断器的功能，并指定了fallbackMethod熔断方法
    @HystrixCommand(fallbackMethod = "hiError")
    public String hiService(String name){
        //调用接口进行消费
        String result = restTemplate.getForObject("http://PRODUCER/hello?name="+name,String.class);
        return result;
    }
    public String hiError(String name) {
        return "hi,"+name+"error!";
    }
}

HystrixObservableCommand command = new HystrixObservableCommand(arg1, arg2);

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@EnableCircuitBreaker
public @interface EnableHystrix {
}

/**
 * Annotation to enable a CircuitBreaker implementation.
 * http://martinfowler.com/bliki/CircuitBreaker.html
 * @author Spencer Gibb
 */
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@Import(EnableCircuitBreakerImportSelector.class)
public @interface EnableCircuitBreaker {
}

/**
 * Import a single circuit breaker implementation Configuration
 * @author Spencer Gibb
 */
@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);
    }
}

/**
 * @author Spencer Gibb
 * @author Christian Dupuis
 * @author Venil Noronha
 */
@Configuration
public class HystrixCircuitBreakerConfiguration {
    @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));
    }
......

/**
 * AspectJ aspect to process methods which annotated with {@link HystrixCommand} annotation.
 */
@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();
    }
 //定义切点，切到 @HystrixCommand标签所在的方法  
  @Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand)")
  public void hystrixCommandAnnotationPointcut() {
  }
@Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCollapser)")
  public void hystrixCollapserAnnotationPointcut() {
  }
   //针对切点：@hystrixCommand切点的处理
    @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);
        //判断方法上不能同时存在@HystrixCommand标签和HystrixCollapser标签
        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 
        HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder);
        ExecutionType executionType = metaHolder.isCollapserAnnotationPresent() ?
                metaHolder.getCollapserExecutionType() : metaHolder.getExecutionType();
        Object result;
        try {
           // 通过CommandExecutor来执行方法
            if (!metaHolder.isObservable()) {
                result = CommandExecutor.execute(invokable, executionType, metaHolder);
            } else {
                result = executeObservable(invokable, executionType, metaHolder);
            }
        } catch (HystrixBadRequestException e) {
            throw e.getCause() != null ? e.getCause() : e;
        } catch (HystrixRuntimeException e) {
            throw hystrixRuntimeExceptionToThrowable(metaHolder, e);
        }
        return result;

   public HystrixInvokable create(MetaHolder metaHolder) {
        HystrixInvokable executable;
        ...省略代码...
            executable = new GenericCommand(HystrixCommandBuilderFactory.getInstance().create(metaHolder));
        }
        return executable;
    }

@ThreadSafe
public class GenericCommand extends AbstractHystrixCommand<Object> {
    private static final Logger LOGGER = LoggerFactory.getLogger(GenericCommand.class);
    public GenericCommand(HystrixCommandBuilder builder) {
        super(builder);
    }
    protected Object run() throws Exception {
        LOGGER.debug("execute command: {}", this.getCommandKey().name());
        return this.process(new AbstractHystrixCommand<Object>.Action() {
            Object execute() {
                return GenericCommand.this.getCommandAction().execute(GenericCommand.this.getExecutionType());
            }
        });
    }
    protected Object getFallback() {
        final CommandAction commandAction = this.getFallbackAction();
        if (commandAction != null) {
            try {
                return this.process(new AbstractHystrixCommand<Object>.Action() {
                    Object execute() {
                        MetaHolder metaHolder = commandAction.getMetaHolder();
                        Object[] args = CommonUtils.createArgsForFallback(metaHolder, GenericCommand.this.getExecutionException());
                        return commandAction.executeWithArgs(metaHolder.getFallbackExecutionType(), args);
                    }
                });
            } catch (Throwable var3) {
                LOGGER.error(FallbackErrorMessageBuilder.create().append(commandAction, var3).build());
                throw new FallbackInvocationException(ExceptionUtils.unwrapCause(var3));
            }
        } else {
            return super.getFallback();
        }
    }
}

public class CommandExecutor {
    public CommandExecutor() {
    }
    public static Object execute(HystrixInvokable invokable, ExecutionType executionType, MetaHolder metaHolder) throws RuntimeException {
        Validate.notNull(invokable);
        Validate.notNull(metaHolder);
        switch(executionType) {
        //异步
        case SYNCHRONOUS:
            return castToExecutable(invokable, executionType).execute();
        //同步
        case ASYNCHRONOUS:
            HystrixExecutable executable = castToExecutable(invokable, executionType);
            if (metaHolder.hasFallbackMethodCommand() && ExecutionType.ASYNCHRONOUS == metaHolder.getFallbackExecutionType()) {
                return new FutureDecorator(executable.queue());
            }
            return executable.queue();
        case OBSERVABLE:
            HystrixObservable observable = castToObservable(invokable);
            return ObservableExecutionMode.EAGER == metaHolder.getObservableExecutionMode() ? observable.observe() : observable.toObservable();
        default:
            throw new RuntimeException("unsupported execution type: " + executionType);
        }
    }
    private static HystrixExecutable castToExecutable(HystrixInvokable invokable, ExecutionType executionType) {
        if (invokable instanceof HystrixExecutable) {
            return (HystrixExecutable)invokable;
        } else {
            throw new RuntimeException("Command should implement " + HystrixExecutable.class.getCanonicalName() + " interface to execute in: " + executionType + " mode");
        }
    }

 public R execute() {
        try {
            return queue().get();
        } catch (Exception e) {
            throw Exceptions.sneakyThrow(decomposeException(e));
        }
    }
--------------
 public Future<R> queue() {
        /*
         * The Future returned by Observable.toBlocking().toFuture() does not implement the
         * interruption of the execution thread when the "mayInterrupt" flag of Future.cancel(boolean) is set to true;
         * thus, to comply with the contract of Future, we must wrap around it.
         */
        final Future<R> delegate = toObservable().toBlocking().toFuture();
        final Future<R> f = new Future<R>() {
            @Override
            public boolean cancel(boolean mayInterruptIfRunning) {
                if (delegate.isCancelled()) {
                    return false;
                }
                if (HystrixCommand.this.getProperties().executionIsolationThreadInterruptOnFutureCancel().get()) {
                    /*
                     * The only valid transition here is false -> true. If there are two futures, say f1 and f2, created by this command
                     * (which is super-weird, but has never been prohibited), and calls to f1.cancel(true) and to f2.cancel(false) are
                     * issued by different threads, it's unclear about what value would be used by the time mayInterruptOnCancel is checked.
                     * The most consistent way to deal with this scenario is to say that if *any* cancellation is invoked with interruption,
                     * than that interruption request cannot be taken back.
                     */
                    interruptOnFutureCancel.compareAndSet(false, mayInterruptIfRunning);
                }
                final boolean res = delegate.cancel(interruptOnFutureCancel.get());
                if (!isExecutionComplete() && interruptOnFutureCancel.get()) {
                    final Thread t = executionThread.get();
                    if (t != null && !t.equals(Thread.currentThread())) {
                        t.interrupt();
                    }
                }
                return res;
            }
....省略...

private static HystrixCircuitBreaker initCircuitBreaker(boolean enabled, HystrixCircuitBreaker fromConstructor, HystrixCommandKey commandKey...) {
    // 如果启用了熔断器
    if (enabled) {
        // 若commandKey没有对应的CircuitBreaker,则创建
        if (fromConstructor == null) {
            return HystrixCircuitBreaker.Factory.getInstance(commandKey, groupKey, properties, metrics);
        } else {
            // 如果有则返回现有的
            return fromConstructor;
        }
    } else {
        return new NoOpCircuitBreaker();
    }
}

 public static class Factory {
 // 维护一个ConcurrentHashMap保存熔断器
        private static ConcurrentHashMap<String, HystrixCircuitBreaker> circuitBreakersByCommand = new ConcurrentHashMap();
        public Factory() {
        }
        public static HystrixCircuitBreaker getInstance(HystrixCommandKey key, HystrixCommandGroupKey group, HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
            HystrixCircuitBreaker previouslyCached = (HystrixCircuitBreaker)circuitBreakersByCommand.get(key.name());
            // 判断ConcurrentHashMap中是否存有对应key的熔断器对象
            if (previouslyCached != null) {
                return previouslyCached;
            } else {
            // 实例化并添加到ConcurrentHashMap里
                HystrixCircuitBreaker cbForCommand = (HystrixCircuitBreaker)circuitBreakersByCommand.putIfAbsent(key.name(), new HystrixCircuitBreaker.HystrixCircuitBreakerImpl(key, group, properties, metrics));
                return cbForCommand == null ? (HystrixCircuitBreaker)circuitBreakersByCommand.get(key.name()) : cbForCommand;
            }
        }
        public static HystrixCircuitBreaker getInstance(HystrixCommandKey key) {
            return (HystrixCircuitBreaker)circuitBreakersByCommand.get(key.name());
        }
        static void reset() {
            circuitBreakersByCommand.clear();
        }
    }
}

 public static class HystrixCircuitBreakerImpl implements HystrixCircuitBreaker {
        // 熔断器的属性常量类
        private final HystrixCommandProperties properties;
       // 熔断器的熔断指标类
        private final HystrixCommandMetrics metrics;
        // 定义一个原子类型的布尔值状态
        private AtomicBoolean circuitOpen = new AtomicBoolean(false);
        private AtomicLong circuitOpenedOrLastTestedTime = new AtomicLong();
        protected HystrixCircuitBreakerImpl(HystrixCommandKey key, HystrixCommandGroupKey commandGroup, HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
            this.properties = properties;
            this.metrics = metrics;
        }
// 重置统计数据
        public void markSuccess() {
            if (this.circuitOpen.get() && this.circuitOpen.compareAndSet(true, false)) {
                this.metrics.resetStream();
            }
        }
// 是否放行请求
        public boolean allowRequest() {
        // 熔断器开启则不放行
            if ((Boolean)this.properties.circuitBreakerForceOpen().get()) {
                return false;
            } else if ((Boolean)this.properties.circuitBreakerForceClosed().get()) {
                this.isOpen();
                return true;
            } else {
                return !this.isOpen() || this.allowSingleTest();
            }
        }
// 部分放行
        public boolean allowSingleTest() {
            long timeCircuitOpenedOrWasLastTested = this.circuitOpenedOrLastTestedTime.get();
            return this.circuitOpen.get() && System.currentTimeMillis() > timeCircuitOpenedOrWasLastTested + (long)(Integer)this.properties.circuitBreakerSleepWindowInMilliseconds().get() && this.circuitOpenedOrLastTestedTime.compareAndSet(timeCircuitOpenedOrWasLastTested, System.currentTimeMillis());
        }
// 判断熔断器是否开启
        public boolean isOpen() {
            if (this.circuitOpen.get()) {
                return true;
            } else {
               // 根据采集到的数据进行判断是否开启
                HealthCounts health = this.metrics.getHealthCounts();
                if (health.getTotalRequests() < (long)(Integer)this.properties.circuitBreakerRequestVolumeThreshold().get()) {
                    return false;
                } else if (health.getErrorPercentage() < (Integer)this.properties.circuitBreakerErrorThresholdPercentage().get()) {
                    return false;
                } else if (this.circuitOpen.compareAndSet(false, true)) {
                    this.circuitOpenedOrLastTestedTime.set(System.currentTimeMillis());
                    return true;
                } else {
                    return true;
                }
            }
        }
    }

protected HystrixCircuitBreakerImpl(HystrixCommandKey key, HystrixCommandGroupKey commandGroup, final HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
    this.properties = properties;
    // 这是Command中的metrics对象,metrics对象也是commandKey维度的
    this.metrics = metrics;
    // !!!重点:订阅事件流
    Subscription s = subscribeToStream();
    activeSubscription.set(s);
}
// 订阅事件流, 前面打的比方: 小溪汇成的河, 各事件以结构化数据汇入了Stream中
private Subscription subscribeToStream() {
    // HealthCountsStream是重点，下面会分析
    return metrics.getHealthCountsStream()
            .observe()
            // 利用数据统计的结果HealthCounts, 实现熔断器
            .subscribe(new Subscriber<HealthCounts>() {
                @Override
                public void onCompleted() {}
                @Override
                public void onError(Throwable e) {}
                @Override
                public void onNext(HealthCounts hc) {
                    // 检查是否达到最小请求数,默认20个; 未达到的话即使请求全部失败也不会熔断
                    if (hc.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {
                        // 啥也不做
                    } else {
                        // 错误百分比未达到设定的阀值
                        if (hc.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) {
                        } else {
                            // 错误率过高, 进行熔断
                            if (status.compareAndSet(Status.CLOSED, Status.OPEN)) {
                                circuitOpened.set(System.currentTimeMillis());
                            }
                        }
                    }
                }
            });
}

public static class HealthCounts {
    // rolling window 中请求总数量
    private final long totalCount;
    // 错误请求数(failure + success + timeout + threadPoolRejected + semaphoreRejected)
    private final long errorCount;
    // 错误率
    private final int errorPercentage;
}

// 是否允许执行
public boolean attemptExecution() {
    // 熔断器配置了强制打开, 不允许执行命令
    if (properties.circuitBreakerForceOpen().get()) {
        return false;
    }
    // 熔断器配置了强制关闭, 允许执行
    if (properties.circuitBreakerForceClosed().get()) {
        return true;
    }
    // AtomicLong circuitOpened, -1是表示熔断器未打开
    if (circuitOpened.get() == -1) {
        return true;
    } else {
        // 熔断后,会拒绝所有命令一段时间(默认5s), 称为sleepWindow
        if (isAfterSleepWindow()) {
            // 过了sleepWindow后,将熔断器设置为"HALF_OPEN",允许第一个请求过去
            if (status.compareAndSet(Status.OPEN, Status.HALF_OPEN)) {
                return true;
            } else {
                return false;
            }
        } else {
            return false;
        }
    }
}

public void markSuccess() {
    // 如果是HALF_OPEN状态,则关闭熔断器
    if (status.compareAndSet(Status.HALF_OPEN, Status.CLOSED)) {
        // 重新开始统计metrics,抛弃所有原先的metrics信息
        metrics.resetStream();
        Subscription previousSubscription = activeSubscription.get();
        if (previousSubscription != null) {
            previousSubscription.unsubscribe();
        }
        Subscription newSubscription = subscribeToStream();
        activeSubscription.set(newSubscription);
        // circuitOpened设置为-1表示关闭熔断器
        circuitOpened.set(-1L);
    }
}

final Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() {
    @Override
    public Observable<R> call(Throwable t) {
        circuitBreaker.markNonSuccess();
        Exception e = getExceptionFromThrowable(t);
        executionResult = executionResult.setExecutionException(e);
        // 线程池拒绝
        if (e instanceof RejectedExecutionException) {
            return handleThreadPoolRejectionViaFallback(e);
        // 超时
        } else if (t instanceof HystrixTimeoutException) {
            return handleTimeoutViaFallback();
        // Bad Request    
        } else if (t instanceof HystrixBadRequestException) {
            return handleBadRequestByEmittingError(e);
        } else {
            if (e instanceof HystrixBadRequestException) {
                eventNotifier.markEvent(HystrixEventType.BAD_REQUEST, commandKey);
                return Observable.error(e);
            }
            return handleFailureViaFallback(e);
        }
    }
};