文档章节

聊聊HystrixThreadPool

go4it
 go4it
发布于 06/25 23:10
字数 1202
阅读 9
收藏 1
点赞 0
评论 0

本文主要研究一下HystrixThreadPool

HystrixThreadPool

hystrix-core-1.5.12-sources.jar!/com/netflix/hystrix/HystrixThreadPool.java

/**
 * ThreadPool used to executed {@link HystrixCommand#run()} on separate threads when configured to do so with {@link HystrixCommandProperties#executionIsolationStrategy()}.
 * <p>
 * Typically each {@link HystrixCommandGroupKey} has its own thread-pool so that any one group of commands can not starve others from being able to run.
 * <p>
 * A {@link HystrixCommand} can be configured with a thread-pool explicitly by injecting a {@link HystrixThreadPoolKey} or via the
 * {@link HystrixCommandProperties#executionIsolationThreadPoolKeyOverride()} otherwise it
 * will derive a {@link HystrixThreadPoolKey} from the injected {@link HystrixCommandGroupKey}.
 * <p>
 * The pool should be sized large enough to handle normal healthy traffic but small enough that it will constrain concurrent execution if backend calls become latent.
 * <p>
 * For more information see the Github Wiki: https://github.com/Netflix/Hystrix/wiki/Configuration#wiki-ThreadPool and https://github.com/Netflix/Hystrix/wiki/How-it-Works#wiki-Isolation
 */
public interface HystrixThreadPool {

    /**
     * Implementation of {@link ThreadPoolExecutor}.
     *
     * @return ThreadPoolExecutor
     */
    public ExecutorService getExecutor();

    public Scheduler getScheduler();

    public Scheduler getScheduler(Func0<Boolean> shouldInterruptThread);

    /**
     * Mark when a thread begins executing a command.
     */
    public void markThreadExecution();

    /**
     * Mark when a thread completes executing a command.
     */
    public void markThreadCompletion();

    /**
     * Mark when a command gets rejected from the threadpool
     */
    public void markThreadRejection();

    /**
     * Whether the queue will allow adding an item to it.
     * <p>
     * This allows dynamic control of the max queueSize versus whatever the actual max queueSize is so that dynamic changes can be done via property changes rather than needing an app
     * restart to adjust when commands should be rejected from queuing up.
     *
     * @return boolean whether there is space on the queue
     */
    public boolean isQueueSpaceAvailable();

    //......
}

这个接口主要是定义getExecutor、getScheduler方法,以及一系列的mark方法(markThreadExecution、markThreadCompletion、markThreadRejection)

HystrixThreadPoolDefault

    /**
     * @ExcludeFromJavadoc
     * @ThreadSafe
     */
    /* package */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);
        }

        @Override
        public ThreadPoolExecutor getExecutor() {
            touchConfig();
            return threadPool;
        }

        @Override
        public Scheduler getScheduler() {
            //by default, interrupt underlying threads on timeout
            return getScheduler(new Func0<Boolean>() {
                @Override
                public Boolean call() {
                    return true;
                }
            });
        }

        @Override
        public Scheduler getScheduler(Func0<Boolean> shouldInterruptThread) {
            touchConfig();
            return new HystrixContextScheduler(HystrixPlugins.getInstance().getConcurrencyStrategy(), this, shouldInterruptThread);
        }

        // allow us to change things via fast-properties by setting it each time
        private void touchConfig() {
            final int dynamicCoreSize = properties.coreSize().get();
            final int configuredMaximumSize = properties.maximumSize().get();
            int dynamicMaximumSize = properties.actualMaximumSize();
            final boolean allowSizesToDiverge = properties.getAllowMaximumSizeToDivergeFromCoreSize().get();
            boolean maxTooLow = false;

            if (allowSizesToDiverge && configuredMaximumSize < dynamicCoreSize) {
                //if user sets maximum < core (or defaults get us there), we need to maintain invariant of core <= maximum
                dynamicMaximumSize = dynamicCoreSize;
                maxTooLow = true;
            }

            // In JDK 6, setCorePoolSize and setMaximumPoolSize will execute a lock operation. Avoid them if the pool size is not changed.
            if (threadPool.getCorePoolSize() != dynamicCoreSize || (allowSizesToDiverge && threadPool.getMaximumPoolSize() != dynamicMaximumSize)) {
                if (maxTooLow) {
                    logger.error("Hystrix ThreadPool configuration for : " + metrics.getThreadPoolKey().name() + " is trying to set coreSize = " +
                            dynamicCoreSize + " and maximumSize = " + configuredMaximumSize + ".  Maximum size will be set to " +
                            dynamicMaximumSize + ", the coreSize value, since it must be equal to or greater than the coreSize value");
                }
                threadPool.setCorePoolSize(dynamicCoreSize);
                threadPool.setMaximumPoolSize(dynamicMaximumSize);
            }

            threadPool.setKeepAliveTime(properties.keepAliveTimeMinutes().get(), TimeUnit.MINUTES);
        }

        @Override
        public void markThreadExecution() {
            metrics.markThreadExecution();
        }

        @Override
        public void markThreadCompletion() {
            metrics.markThreadCompletion();
        }

        @Override
        public void markThreadRejection() {
            metrics.markThreadRejection();
        }

        /**
         * Whether the threadpool queue has space available according to the <code>queueSizeRejectionThreshold</code> settings.
         *
         * Note that the <code>queueSize</code> is an final instance variable on HystrixThreadPoolDefault, and not looked up dynamically.
         * The data structure is static, so this does not make sense as a dynamic lookup.
         * The <code>queueSizeRejectionThreshold</code> can be dynamic (up to <code>queueSize</code>), so that should
         * still get checked on each invocation.
         * <p>
         * If a SynchronousQueue implementation is used (<code>maxQueueSize</code> <= 0), it always returns 0 as the size so this would always return true.
         */
        @Override
        public boolean isQueueSpaceAvailable() {
            if (queueSize <= 0) {
                // we don't have a queue so we won't look for space but instead
                // let the thread-pool reject or not
                return true;
            } else {
                return threadPool.getQueue().size() < properties.queueSizeRejectionThreshold().get();
            }
        }

    }
  • getExecutor方法返回的是HystrixThreadPoolMetrics的ThreadPoolExecutor,只是每次get的时候会touch下config,刷新下threadPool的配置
  • getScheduler方法这里用get不是太恰当,因为里头是new了一个HystrixContextScheduler
  • mark方法主要是委托给HystrixThreadPoolMetrics

HystrixThreadPoolProperties

hystrix-core-1.5.12-sources.jar!/com/netflix/hystrix/HystrixThreadPoolProperties.java

/**
 * Properties for instances of {@link HystrixThreadPool}.
 * <p>
 * Default implementation of methods uses Archaius (https://github.com/Netflix/archaius)
 *
 * Note a change in behavior in 1.5.7.  Prior to that version, the configuration for 'coreSize' was used to control
 * both coreSize and maximumSize.  This is a fixed-size threadpool that can never give up an unused thread.  In 1.5.7+,
 * the values can diverge, and if you set coreSize < maximumSize, threads can be given up (subject to the keep-alive
 * time)
 *
 * It is OK to leave maximumSize unset using any version of Hystrix.  If you do, then maximum size will default to
 * core size and you'll have a fixed-size threadpool.
 *
 * If you accidentally set maximumSize < coreSize, then maximum will be raised to coreSize
 * (this prioritizes keeping extra threads around rather than inducing threadpool rejections)
 */
public abstract class HystrixThreadPoolProperties {

    /* defaults */
    static int default_coreSize = 10;            // core size of thread pool
    static int default_maximumSize = 10;         // maximum size of thread pool
    static int default_keepAliveTimeMinutes = 1; // minutes to keep a thread alive
    static int default_maxQueueSize = -1;        // size of queue (this can't be dynamically changed so we use 'queueSizeRejectionThreshold' to artificially limit and reject)
                                                 // -1 turns it off and makes us use SynchronousQueue
    static boolean default_allow_maximum_size_to_diverge_from_core_size = false; //should the maximumSize config value get read and used in configuring the threadPool
                                                                                 //turning this on should be a conscious decision by the user, so we default it to false

    static int default_queueSizeRejectionThreshold = 5; // number of items in queue
    static int default_threadPoolRollingNumberStatisticalWindow = 10000; // milliseconds for rolling number
    static int default_threadPoolRollingNumberStatisticalWindowBuckets = 10; // number of buckets in rolling number (10 1-second buckets)

    private final HystrixProperty<Integer> corePoolSize;
    private final HystrixProperty<Integer> maximumPoolSize;
    private final HystrixProperty<Integer> keepAliveTime;
    private final HystrixProperty<Integer> maxQueueSize;
    private final HystrixProperty<Integer> queueSizeRejectionThreshold;
    private final HystrixProperty<Boolean> allowMaximumSizeToDivergeFromCoreSize;

    private final HystrixProperty<Integer> threadPoolRollingNumberStatisticalWindowInMilliseconds;
    private final HystrixProperty<Integer> threadPoolRollingNumberStatisticalWindowBuckets;
    //......
}

主要是配置corePoolSize、maximumPoolSize、keepAliveTime、maxQueueSize、queueSizeRejectionThreshold、allowMaximumSizeToDivergeFromCoreSize、threadPoolRollingNumberStatisticalWindowInMilliseconds、threadPoolRollingNumberStatisticalWindowBuckets

HystrixThreadPoolMetrics

hystrix-core-1.5.12-sources.jar!/com/netflix/hystrix/HystrixThreadPoolMetrics.java

/**
 * Used by {@link HystrixThreadPool} to record metrics.
 */
public class HystrixThreadPoolMetrics extends HystrixMetrics {
    //......
    /**
     * Invoked each time a thread is executed.
     */
    public void markThreadExecution() {
        concurrentExecutionCount.incrementAndGet();
    }
    
    /**
     * Invoked each time a thread completes.
     */
    public void markThreadCompletion() {
        concurrentExecutionCount.decrementAndGet();
    }
    
    /**
     * Invoked each time a command is rejected from the thread-pool
     */
    public void markThreadRejection() {
        concurrentExecutionCount.decrementAndGet();
    }
    //......
}

这几个mark方法是相应的计数。

小结

HystrixThreadPool主要是定义getExecutor、getScheduler方法,以及一系列的mark方法。这些mark方法主要是做一些指标的count统计。

doc

© 著作权归作者所有

共有 人打赏支持
go4it
粉丝 51
博文 674
码字总数 471634
作品 0
深圳
聊聊hystrix的queueSizeRejectionThreshold参数

序 本文主要研究一下hystrix的queueSizeRejectionThreshold参数 HystrixThreadPoolProperties hystrix-core-1.5.12-sources.jar!/com/netflix/hystrix/HystrixThreadPoolProperties.java 设计......

go4it
07/04
0
0
聊聊HystrixPropertiesStrategy

序 本文主要研究一下HystrixPropertiesStrategy HystrixPropertiesStrategy hystrix-core-1.5.12-sources.jar!/com/netflix/hystrix/strategy/properties/HystrixPropertiesStrategy.java 提......

go4it
07/02
0
0
聊聊HystrixMetricsPublisher

序 本文主要研究一下HystrixMetricsPublisher HystrixMetricsPublisher hystrix-core-1.5.12-sources.jar!/com/netflix/hystrix/strategy/metrics/HystrixMetricsPublisher.java getMetrics......

go4it
07/01
0
0
聊聊HystrixConcurrencyStrategy

序 本文主要研究一下HystrixConcurrencyStrategy HystrixConcurrencyStrategy hystrix-core-1.5.12-sources.jar!/com/netflix/hystrix/strategy/concurrency/HystrixConcurrencyStrategy.ja......

go4it
06/30
0
0
教程:一起学习Hystrix--Hystrix命令名称、分组、线程池

目录 Hystrix本系列 命令名称 命令分组 命令线程池 惊喜 Hystrix本系列 以下为博主写Hystrix系列的文章列表 点击查看 Hystrix入门 点击查看 Hystrix命令执行 点击查看 Hystrix处理异常机制(...

java_龙
05/22
0
0
聊聊HystrixEventStream

序 本文主要研究一下HystrixEventStream HystrixEventStream hystrix-core-1.5.12-sources.jar!/com/netflix/hystrix/metric/HystrixEventStream.java 这个接口定义了一个observe方法,返回的......

go4it
07/05
0
0
聊聊并发系列_Index

聊聊并发系列 聊聊并发(一)深入分析Volatile的实现原理 聊聊并发(二)Java SE1.6中的Synchronized 聊聊并发(三)Java线程池的分析和使用 聊聊并发(四)深入分析ConcurrentHashMap 聊聊并...

陶邦仁
2016/01/04
450
0
自定义处理 hystrix 历史监控数据

在 spring cloud 中引入 hystrix 使我们得以在内部依赖的某个服务异常时能够快速失败和做降级处理,而不至于级联影响到本身对其他接口的服务。 但是 hystrix 的配置还是蛮的,比如线程池隔离...

holysu
05/17
0
0
聊聊远程通信_Index

聊聊远程通信 Java远程通讯技术及原理分析 聊聊Socket、TCP/IP、HTTP、FTP及网络编程 RMI原理及实现 RPC原理及实现 轻量级分布式 RPC 框架 使用 RMI + ZooKeeper 实现远程调用框架 聊聊同步、...

陶邦仁
2016/02/23
1K
0
腾讯—iOS社招面试

丢了几份简历给腾讯的iOS,很多都给标为不合适,倒是有个MIG部门让我去面。 约了下午3点,在大族大厦12楼一面(面试官看着好实诚): 面试内容: 高工面 1.property本质-->property关键字 2....

gdxz110
2016/03/23
821
2

没有更多内容

加载失败,请刷新页面

加载更多

下一页

SpringBoot | 第十章:Swagger2的集成和使用

前言 前一章节介绍了mybatisPlus的集成和简单使用,本章节开始接着上一章节的用户表,进行Swagger2的集成。现在都奉行前后端分离开发和微服务大行其道,分微服务及前后端分离后,前后端开发的...

oKong
今天
2
0
Python 最小二乘法 拟合 二次曲线

Python 二次拟合 随机生成数据,并且加上噪声干扰 构造需要拟合的函数形式,使用最小二乘法进行拟合 输出拟合后的参数 将拟合后的函数与原始数据绘图后进行对比 import numpy as npimport...

阿豪boy
今天
1
0
云拿 无人便利店

附近(上海市-航南路)开了家无人便利店.特意进去体验了一下.下面把自己看到的跟大家分享下. 经得现场工作人员同意后拍了几张照片.从外面看是这样.店门口的指导里强调:不要一次扫码多个人进入....

周翔
昨天
1
0
Java设计模式学习之工厂模式

在Java(或者叫做面向对象语言)的世界中,工厂模式被广泛应用于项目中,也许你并没有听说过,不过也许你已经在使用了。 简单来说,工厂模式的出现源于增加程序序的可扩展性,降低耦合度。之...

路小磊
昨天
161
1
npm profile 新功能介绍

转载地址 npm profile 新功能介绍 npm新版本新推来一个功能,npm profile,这个可以更改自己简介信息的命令,以后可以不用去登录网站来修改自己的简介了 具体的这个功能的支持大概是在6这个版...

durban
昨天
1
0
Serial2Ethernet Bi-redirection

Serial Tool Serial Tool is a utility for developing serial communications, custom protocols or device testing. You can set up bytes to send accordingly to your protocol and save......

zungyiu
昨天
1
0
python里求解物理学上的双弹簧质能系统

物理的模型如下: 在这个系统里有两个物体,它们的质量分别是m1和m2,被两个弹簧连接在一起,伸缩系统为k1和k2,左端固定。假定没有外力时,两个弹簧的长度为L1和L2。 由于两物体有重力,那么...

wangxuwei
昨天
0
0
apolloxlua 介绍

##项目介绍 apolloxlua 目前支持javascript到lua的翻译。可以在openresty和luajit里使用。这个工具分为两种模式, 一种是web模式,可以通过网页使用。另外一种是tool模式, 通常作为大规模翻...

钟元OSS
昨天
2
0
Mybatis入门

简介: 定义:Mybatis是一个支持普通SQL查询、存储过程和高级映射的持久层框架。 途径:MyBatis通过XML文件或者注解的形式配置映射,实现数据库查询。 特性:动态SQL语句。 文件结构:Mybat...

霍淇滨
昨天
2
0
开发技术瓶颈期,如何突破

前言 读书、学习的那些事情,以前我也陆续叨叨了不少,但总觉得 “学习方法” 就是一个永远在路上的话题。个人的能力、经验积累与习惯方法不尽相同,而且一篇文章甚至一本书都很难将学习方法...

_小迷糊
昨天
1
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部