Hystrix超时机制

简介: 开篇 Hystrix的超时检测本质上通过启动单独线程去检测的,线程的启动的时间刚好就是任务超时的时间,本质上就是这么个简单的逻辑。 Hystrix超时后会抛出一个HystrixTimeoutException的异常。

开篇

 Hystrix的超时检测本质上通过启动单独线程去检测的,线程的启动的时间刚好就是任务超时的时间,本质上就是这么个简单的逻辑。
 Hystrix超时后会抛出一个HystrixTimeoutException的异常。


超时检测逻辑

 Hystrix的超时包括注册过程和执行过程两个,注册过程如下:

  • 执行lift(new HystrixObservableTimeoutOperator<R>(_cmd))关联超时检测任务。

  • 在HystrixObservableTimeoutOperator类中,new TimerListener()负责创建检测任务,HystrixTimer.getInstance().addTimerListener(listener)负责关联定时任务。

  • 在HystrixObservableTimeoutOperator类中,addTimerListener通过java的定时任务服务scheduleAtFixedRate在延迟超时时间后执行



 Hystrix的超时执行过程如下:

  • 在超时后执行 listener.tick()方法后执行类TimerListener的tick方法

  • 在TimerListener类的tick方法中执行timeoutRunnable.run()后执行HystrixContextRunnable的run方法

  • 在HystrixContextRunnable类run方法中执行child.onError(new HystrixTimeoutException())实现超时。

executeCommandWithSpecifiedIsolation(_cmd).lift(new HystrixObservableTimeoutOperator<R>(_cmd));



private static class HystrixObservableTimeoutOperator<R> implements Operator<R, R> {

        final AbstractCommand<R> originalCommand;

        public HystrixObservableTimeoutOperator(final AbstractCommand<R> originalCommand) {
            this.originalCommand = originalCommand;
        }

        @Override
        public Subscriber<? super R> call(final Subscriber<? super R> child) {
            final CompositeSubscription s = new CompositeSubscription();
            // if the child unsubscribes we unsubscribe our parent as well
            child.add(s);

            //capture the HystrixRequestContext upfront so that we can use it in the timeout thread later
            final HystrixRequestContext hystrixRequestContext = HystrixRequestContext.getContextForCurrentThread();

            TimerListener listener = new TimerListener() {

                @Override
                public void tick() {
                  if (originalCommand.isCommandTimedOut.compareAndSet(TimedOutStatus.NOT_EXECUTED, TimedOutStatus.TIMED_OUT)) {
                        // report timeout failure
                        originalCommand.eventNotifier.markEvent(HystrixEventType.TIMEOUT, originalCommand.commandKey);

                        // shut down the original request
                        s.unsubscribe();

                        final HystrixContextRunnable timeoutRunnable = new HystrixContextRunnable(
                                originalCommand.concurrencyStrategy, hystrixRequestContext, new Runnable() {

                            @Override
                            public void run() {
                                child.onError(new HystrixTimeoutException());
                            }
                        });

                        timeoutRunnable.run();
                    }
                }

                @Override
                public int getIntervalTimeInMilliseconds() {
                    return originalCommand.properties.executionTimeoutInMilliseconds().get();
                }
            };

            final Reference<TimerListener> tl = HystrixTimer.getInstance().addTimerListener(listener);

            // set externally so execute/queue can see this
            originalCommand.timeoutTimer.set(tl);

            /**
             * If this subscriber receives values it means the parent succeeded/completed
             */
            Subscriber<R> parent = new Subscriber<R>() {

                @Override
                public void onCompleted() {
                    if (isNotTimedOut()) {
                        // stop timer and pass notification through
                        tl.clear();
                        child.onCompleted();
                    }
                }

                @Override
                public void onError(Throwable e) {
                    if (isNotTimedOut()) {
                        // stop timer and pass notification through
                        tl.clear();
                        child.onError(e);
                    }
                }

                @Override
                public void onNext(R v) {
                    if (isNotTimedOut()) {
                        child.onNext(v);
                    }
                }

                private boolean isNotTimedOut() {
                    // if already marked COMPLETED (by onNext) or succeeds in setting to COMPLETED
                    return originalCommand.isCommandTimedOut.get() == TimedOutStatus.COMPLETED ||
                            originalCommand.isCommandTimedOut.compareAndSet(TimedOutStatus.NOT_EXECUTED, 
                                                                             TimedOutStatus.COMPLETED);
                }

            };

            // if s is unsubscribed we want to unsubscribe the parent
            s.add(parent);

            return parent;
        }
    }
    public Reference<TimerListener> addTimerListener(final TimerListener listener) {
        startThreadIfNeeded();
        // add the listener

        Runnable r = new Runnable() {

            @Override
            public void run() {
                try {
                    listener.tick();
                } catch (Exception e) {
                    logger.error("Failed while ticking TimerListener", e);
                }
            }
        };

        //这里直接简单粗暴的scheduleAtFixedRate以超时时间作为周期去判断是否执行完成
        ScheduledFuture<?> f = executor.get().getThreadPool().scheduleAtFixedRate(r, 
                listener.getIntervalTimeInMilliseconds(), listener.getIntervalTimeInMilliseconds(), TimeUnit.MILLISECONDS);
        return new TimerReference(listener, f);
    }
public class HystrixContextRunnable implements Runnable {

    private final Callable<Void> actual;
    private final HystrixRequestContext parentThreadState;

    public HystrixContextRunnable(Runnable actual) {
        this(HystrixPlugins.getInstance().getConcurrencyStrategy(), actual);
    }
    
    public HystrixContextRunnable(HystrixConcurrencyStrategy concurrencyStrategy, final Runnable actual) {
        this(concurrencyStrategy, HystrixRequestContext.getContextForCurrentThread(), actual);
    }

    public HystrixContextRunnable(final HystrixConcurrencyStrategy concurrencyStrategy, 
                                  final HystrixRequestContext hystrixRequestContext, final Runnable actual) {
        this.actual = concurrencyStrategy.wrapCallable(new Callable<Void>() {

            @Override
            public Void call() throws Exception {
                actual.run();
                return null;
            }

        });
        this.parentThreadState = hystrixRequestContext;
    }

    @Override
    public void run() {
        HystrixRequestContext existingState = HystrixRequestContext.getContextForCurrentThread();
        try {
            // set the state of this thread to that of its parent
            HystrixRequestContext.setContextOnCurrentThread(parentThreadState);
            // execute actual Callable with the state of the parent
            try {
                actual.call();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        } finally {
            // restore this thread back to its original state
            HystrixRequestContext.setContextOnCurrentThread(existingState);
        }
    }
}


参考文章

目录
相关文章
|
9月前
Ribbon、Feign、Hystrix超时&重试&熔断问题
在使用Ribbon、Feign、Hystrix组合时,因为配置的问题出现以下现象,让我的大脑CPU烧的不行不行(拿我老家话说就是“脑子ran滴奥”)
107 0
|
8月前
|
安全 Java
Hystrix超时机制为服务接口调用超时提供安全保护
Hystrix超时机制为服务接口调用超时提供安全保护
88 1
|
应用服务中间件
高并发下hystrix熔断超时及concurrent.RejectedExecutionException: Rejected command because thread-pool queueSiz...
高并发下hystrix熔断超时及concurrent.RejectedExecutionException: Rejected command because thread-pool queueSiz...
239 0
|
设计模式 缓存 安全
【Hystrix技术指南】(6)请求合并机制原理分析
【Hystrix技术指南】(6)请求合并机制原理分析
162 0
【Hystrix技术指南】(6)请求合并机制原理分析
|
设计模式 Java 数据中心
【Hystrix技术指南】(3)超时机制的原理和实现
【Hystrix技术指南】(3)超时机制的原理和实现
205 0
|
监控 数据可视化 Java
微服务学习笔记八 Spring Cloud Hystrix容错机制
微服务学习笔记八 Spring Cloud Hystrix容错机制
138 0
|
缓存 Java 应用服务中间件
Hystrix线程池机制的资源隔离在业务中的最佳实践
Hystrix线程池机制的资源隔离在业务中的最佳实践
161 0
|
Java Spring
spring cloud:Edgware.RELEASE版本hystrix超时新坑
作者:菩提树下的杨过 出处:http://yjmyzz.cnblogs.com 本文版权归作者和博客园共有,欢迎转载,但未经作者同意必须保留此段声明,且在文章页面明显位置给出原文连接,否则保留追究法律责任的权利。
1061 0
|
Java Spring 微服务
spring cloud使用hystrix实现断路保护机制
断路保护机制:即容错性,在微服务架构中,系统之间都是相互依赖的,如果一个系统出现了异常,就会出现如下情况: 断路保护机制就是为了解决这种情况。 hystrix的大致原理如下: spring cloud中使用断路保护机制非常简单 1.
1079 0