- 浏览: 1583478 次
- 性别:
- 来自: 杭州
文章分类
最新评论
-
jsrgzhangzhiyong:
关于null值的转换还是感觉不太友好,就像 mapstruct ...
我也造了个轮子:BeanMapping(属性拷贝) -
he037:
a417930422 写道引用使用EPHEMERAL会引出一个 ...
基于zookeeper的分布式lock实现 -
seancheer:
qianshangding 写道首先节点启动后,尝试读取本地的 ...
zookeeper学习记录三(session,watcher,persit机制) -
雪夜归人:
您好,我想咨询一下,开源的canal都能支持mysql的哪些版 ...
Canal BinlogChange(mysql5.6) -
zhoudengyun:
copy 一份做记录,后续学习,请知悉
阿里巴巴开源项目: 基于mysql数据库binlog的增量订阅&消费
背景
前段时间一个项目中因为涉及大量的线程开发,把jdk cocurrent的代码重新再过了一遍。这篇文章中主要是记录一下学习ThreadPoolExecutor过程中容易被人忽略的点,Doug Lea的整个类设计还是非常nice的
正文
先看一副图,描述了ThreadPoolExecutor的工作机制:
整个ThreadPoolExecutor的任务处理有4步操作:
- 第一步,初始的poolSize < corePoolSize,提交的runnable任务,会直接做为new一个Thread的参数,立马执行
- 第二步,当提交的任务数超过了corePoolSize,就进入了第二步操作。会将当前的runable提交到一个block queue中
- 第三步,如果block queue是个有界队列,当队列满了之后就进入了第三步。如果poolSize < maximumPoolsize时,会尝试new 一个Thread的进行救急处理,立马执行对应的runnable任务
- 第四步,如果第三步救急方案也无法处理了,就会走到第四步执行reject操作。
几点说明:(相信这些网上一搜一大把,我这里简单介绍下,为后面做一下铺垫)
- block queue有以下几种实现:
1. ArrayBlockingQueue : 有界的数组队列
2. LinkedBlockingQueue : 可支持有界/无界的队列,使用链表实现
3. PriorityBlockingQueue : 优先队列,可以针对任务排序
4. SynchronousQueue : 队列长度为1的队列,和Array有点区别就是:client thread提交到block queue会是一个阻塞过程,直到有一个worker thread连接上来poll task。
- RejectExecutionHandler是针对任务无法处理时的一些自保护处理:
1. Reject 直接抛出Reject exception
2. Discard 直接忽略该runnable,不可取
3. DiscardOldest 丢弃最早入队列的的任务
4. CallsRun 直接让原先的client thread做为worker线程,进行执行
容易被人忽略的点:
1. pool threads启动后,以后的任务获取都会通过block queue中,获取堆积的runnable task.
所以建议: block size >= corePoolSize ,不然线程池就没任何意义
2. corePoolSize 和 maximumPoolSize的区别, 和大家正常理解的数据库连接池不太一样。
* 据dbcp pool为例,会有minIdle , maxActive配置。minIdle代表是常驻内存中的threads数量,maxActive代表是工作的最大线程数。
* 这里的corePoolSize就是连接池的maxActive的概念,它没有minIdle的概念(每个线程可以设置keepAliveTime,超过多少时间多有任务后销毁线程,默认只会针对maximumPoolSize参数的线程生效,可以设置allowCoreThreadTimeOut=true,就可以对corePoolSize进行idle回收)。
* 这里的maximumPoolSize,是一种救急措施的第一层。当threadPoolExecutor的工作threads存在满负荷,并且block queue队列也满了,这时代表接近崩溃边缘。这时允许临时起一批threads,用来处理runnable,处理完后通过keepAliveTime进行调度回收。
所以建议: maximumPoolSize >= corePoolSize =期望的最大线程数。 (我曾经配置了corePoolSize=1, maximumPoolSize=20, blockqueue为无界队列,最后就成了单线程工作的pool。典型的配置错误)
3. 善用blockqueue和reject组合. 这里要重点推荐下CallsRun的Rejected Handler,从字面意思就是让调用者自己来运行。
我们经常会在线上使用一些线程池做异步处理,比如我前面做的(业务层)异步并行加载技术分析和设计, 将原本串行的请求都变为了并行操作,但过多的并行会增加系统的负载(比如软中断,上下文切换)。所以肯定需要对线程池做一个size限制。但是为了引入异步操作后,避免因在block queue的等待时间过长,所以需要在队列满的时,执行一个callsRun的策略,并行的操作又转为一个串行处理,这样就可以保证尽量少的延迟影响。
所以建议: RejectExecutionHandler = CallsRun , blockqueue size = 2 * poolSize (为啥是2倍poolSize,主要一个考虑就是瞬间高峰处理,允许一个thread等待一个runnable任务)
Btrace容量规划
再提供一个btrace脚本,分析线上的thread pool容量规划是否合理,可以运行时输出poolSize等一些数据。
import static com.sun.btrace.BTraceUtils.addToAggregation;
import static com.sun.btrace.BTraceUtils.field;
import static com.sun.btrace.BTraceUtils.get;
import static com.sun.btrace.BTraceUtils.newAggregation;
import static com.sun.btrace.BTraceUtils.newAggregationKey;
import static com.sun.btrace.BTraceUtils.printAggregation;
import static com.sun.btrace.BTraceUtils.println;
import static com.sun.btrace.BTraceUtils.str;
import static com.sun.btrace.BTraceUtils.strcat;
import java.lang.reflect.Field;
import java.util.concurrent.atomic.AtomicInteger;
import com.sun.btrace.BTraceUtils;
import com.sun.btrace.aggregation.Aggregation;
import com.sun.btrace.aggregation.AggregationFunction;
import com.sun.btrace.aggregation.AggregationKey;
import com.sun.btrace.annotations.BTrace;
import com.sun.btrace.annotations.Kind;
import com.sun.btrace.annotations.Location;
import com.sun.btrace.annotations.OnEvent;
import com.sun.btrace.annotations.OnMethod;
import com.sun.btrace.annotations.OnTimer;
import com.sun.btrace.annotations.Self;
/**
* 并行加载监控
*
* @author jianghang 2011-4-7 下午10:59:53
*/
@BTrace
public class AsyncLoadTracer {
private static AtomicInteger rejecctCount = BTraceUtils.newAtomicInteger(0);
private static Aggregation histogram = newAggregation(AggregationFunction.QUANTIZE);
private static Aggregation average = newAggregation(AggregationFunction.AVERAGE);
private static Aggregation max = newAggregation(AggregationFunction.MAXIMUM);
private static Aggregation min = newAggregation(AggregationFunction.MINIMUM);
private static Aggregation sum = newAggregation(AggregationFunction.SUM);
private static Aggregation count = newAggregation(AggregationFunction.COUNT);
@OnMethod(clazz = "java.util.concurrent.ThreadPoolExecutor", method = "execute", location = @Location(value = Kind.ENTRY))
public static void executeMonitor(@Self Object self) {
Field poolSizeField = field("java.util.concurrent.ThreadPoolExecutor", "poolSize");
Field largestPoolSizeField = field("java.util.concurrent.ThreadPoolExecutor", "largestPoolSize");
Field workQueueField = field("java.util.concurrent.ThreadPoolExecutor", "workQueue");
Field countField = field("java.util.concurrent.ArrayBlockingQueue", "count");
int poolSize = (Integer) get(poolSizeField, self);
int largestPoolSize = (Integer) get(largestPoolSizeField, self);
int queueSize = (Integer) get(countField, get(workQueueField, self));
println(strcat(strcat(strcat(strcat(strcat("poolSize : ", str(poolSize)), " largestPoolSize : "),
str(largestPoolSize)), " queueSize : "), str(queueSize)));
}
@OnMethod(clazz = "java.util.concurrent.ThreadPoolExecutor", method = "reject", location = @Location(value = Kind.ENTRY))
public static void rejectMonitor(@Self Object self) {
String name = str(self);
if (BTraceUtils.startsWith(name, "com.alibaba.pivot.common.asyncload.impl.pool.AsyncLoadThreadPool")) {
BTraceUtils.incrementAndGet(rejecctCount);
}
}
@OnTimer(1000)
public static void rejectPrintln() {
int reject = BTraceUtils.getAndSet(rejecctCount, 0);
println(strcat("reject count in 1000 msec: ", str(reject)));
AggregationKey key = newAggregationKey("rejectCount");
addToAggregation(histogram, key, reject);
addToAggregation(average, key, reject);
addToAggregation(max, key, reject);
addToAggregation(min, key, reject);
addToAggregation(sum, key, reject);
addToAggregation(count, key, reject);
}
@OnEvent
public static void onEvent() {
BTraceUtils.truncateAggregation(histogram, 10);
println("---------------------------------------------");
printAggregation("Count", count);
printAggregation("Min", min);
printAggregation("Max", max);
printAggregation("Average", average);
printAggregation("Sum", sum);
printAggregation("Histogram", histogram);
println("---------------------------------------------");
}
}
运行结果:
poolSize : 1 , largestPoolSize = 10 , queueSize = 10
reject count in 1000 msec: 0
说明:
1. poolSize 代表为当前的线程数
2. largestPoolSize 代表为历史最大的线程数
3. queueSize 代表blockqueue的当前堆积的size
4. reject count 代表在1000ms内的被reject的数量
最后
这是我对ThreadPoolExecutor使用过程中的一些经验总结,希望能对大家有所帮助,如有描述不对的地方欢迎拍砖。
背景
前段时间一个项目中因为涉及大量的线程开发,把jdk cocurrent的代码重新再过了一遍。这篇文章中主要是记录一下学习ThreadPoolExecutor过程中容易被人忽略的点,Doug Lea的整个类设计还是非常nice的
正文
先看一副图,描述了ThreadPoolExecutor的工作机制:
整个ThreadPoolExecutor的任务处理有4步操作:
- 第一步,初始的poolSize < corePoolSize,提交的runnable任务,会直接做为new一个Thread的参数,立马执行
- 第二步,当提交的任务数超过了corePoolSize,就进入了第二步操作。会将当前的runable提交到一个block queue中
- 第三步,如果block queue是个有界队列,当队列满了之后就进入了第三步。如果poolSize < maximumPoolsize时,会尝试new 一个Thread的进行救急处理,立马执行对应的runnable任务
- 第四步,如果第三步救急方案也无法处理了,就会走到第四步执行reject操作。
几点说明:(相信这些网上一搜一大把,我这里简单介绍下,为后面做一下铺垫)
- block queue有以下几种实现:
1. ArrayBlockingQueue : 有界的数组队列
2. LinkedBlockingQueue : 可支持有界/无界的队列,使用链表实现
3. PriorityBlockingQueue : 优先队列,可以针对任务排序
4. SynchronousQueue : 队列长度为1的队列,和Array有点区别就是:client thread提交到block queue会是一个阻塞过程,直到有一个worker thread连接上来poll task。 - RejectExecutionHandler是针对任务无法处理时的一些自保护处理:
1. Reject 直接抛出Reject exception
2. Discard 直接忽略该runnable,不可取
3. DiscardOldest 丢弃最早入队列的的任务
4. CallsRun 直接让原先的client thread做为worker线程,进行执行
容易被人忽略的点:
1. pool threads启动后,以后的任务获取都会通过block queue中,获取堆积的runnable task.
所以建议: block size >= corePoolSize ,不然线程池就没任何意义
2. corePoolSize 和 maximumPoolSize的区别, 和大家正常理解的数据库连接池不太一样。
* 据dbcp pool为例,会有minIdle , maxActive配置。minIdle代表是常驻内存中的threads数量,maxActive代表是工作的最大线程数。
* 这里的corePoolSize就是连接池的maxActive的概念,它没有minIdle的概念(每个线程可以设置keepAliveTime,超过多少时间多有任务后销毁线程,默认只会针对maximumPoolSize参数的线程生效,可以设置allowCoreThreadTimeOut=true,就可以对corePoolSize进行idle回收)。
* 这里的maximumPoolSize,是一种救急措施的第一层。当threadPoolExecutor的工作threads存在满负荷,并且block queue队列也满了,这时代表接近崩溃边缘。这时允许临时起一批threads,用来处理runnable,处理完后通过keepAliveTime进行调度回收。
所以建议: maximumPoolSize >= corePoolSize =期望的最大线程数。 (我曾经配置了corePoolSize=1, maximumPoolSize=20, blockqueue为无界队列,最后就成了单线程工作的pool。典型的配置错误)
3. 善用blockqueue和reject组合. 这里要重点推荐下CallsRun的Rejected Handler,从字面意思就是让调用者自己来运行。
我们经常会在线上使用一些线程池做异步处理,比如我前面做的(业务层)异步并行加载技术分析和设计, 将原本串行的请求都变为了并行操作,但过多的并行会增加系统的负载(比如软中断,上下文切换)。所以肯定需要对线程池做一个size限制。但是为了引入异步操作后,避免因在block queue的等待时间过长,所以需要在队列满的时,执行一个callsRun的策略,并行的操作又转为一个串行处理,这样就可以保证尽量少的延迟影响。
所以建议: RejectExecutionHandler = CallsRun , blockqueue size = 2 * poolSize (为啥是2倍poolSize,主要一个考虑就是瞬间高峰处理,允许一个thread等待一个runnable任务)
Btrace容量规划
再提供一个btrace脚本,分析线上的thread pool容量规划是否合理,可以运行时输出poolSize等一些数据。
import static com.sun.btrace.BTraceUtils.addToAggregation; import static com.sun.btrace.BTraceUtils.field; import static com.sun.btrace.BTraceUtils.get; import static com.sun.btrace.BTraceUtils.newAggregation; import static com.sun.btrace.BTraceUtils.newAggregationKey; import static com.sun.btrace.BTraceUtils.printAggregation; import static com.sun.btrace.BTraceUtils.println; import static com.sun.btrace.BTraceUtils.str; import static com.sun.btrace.BTraceUtils.strcat; import java.lang.reflect.Field; import java.util.concurrent.atomic.AtomicInteger; import com.sun.btrace.BTraceUtils; import com.sun.btrace.aggregation.Aggregation; import com.sun.btrace.aggregation.AggregationFunction; import com.sun.btrace.aggregation.AggregationKey; import com.sun.btrace.annotations.BTrace; import com.sun.btrace.annotations.Kind; import com.sun.btrace.annotations.Location; import com.sun.btrace.annotations.OnEvent; import com.sun.btrace.annotations.OnMethod; import com.sun.btrace.annotations.OnTimer; import com.sun.btrace.annotations.Self; /** * 并行加载监控 * * @author jianghang 2011-4-7 下午10:59:53 */ @BTrace public class AsyncLoadTracer { private static AtomicInteger rejecctCount = BTraceUtils.newAtomicInteger(0); private static Aggregation histogram = newAggregation(AggregationFunction.QUANTIZE); private static Aggregation average = newAggregation(AggregationFunction.AVERAGE); private static Aggregation max = newAggregation(AggregationFunction.MAXIMUM); private static Aggregation min = newAggregation(AggregationFunction.MINIMUM); private static Aggregation sum = newAggregation(AggregationFunction.SUM); private static Aggregation count = newAggregation(AggregationFunction.COUNT); @OnMethod(clazz = "java.util.concurrent.ThreadPoolExecutor", method = "execute", location = @Location(value = Kind.ENTRY)) public static void executeMonitor(@Self Object self) { Field poolSizeField = field("java.util.concurrent.ThreadPoolExecutor", "poolSize"); Field largestPoolSizeField = field("java.util.concurrent.ThreadPoolExecutor", "largestPoolSize"); Field workQueueField = field("java.util.concurrent.ThreadPoolExecutor", "workQueue"); Field countField = field("java.util.concurrent.ArrayBlockingQueue", "count"); int poolSize = (Integer) get(poolSizeField, self); int largestPoolSize = (Integer) get(largestPoolSizeField, self); int queueSize = (Integer) get(countField, get(workQueueField, self)); println(strcat(strcat(strcat(strcat(strcat("poolSize : ", str(poolSize)), " largestPoolSize : "), str(largestPoolSize)), " queueSize : "), str(queueSize))); } @OnMethod(clazz = "java.util.concurrent.ThreadPoolExecutor", method = "reject", location = @Location(value = Kind.ENTRY)) public static void rejectMonitor(@Self Object self) { String name = str(self); if (BTraceUtils.startsWith(name, "com.alibaba.pivot.common.asyncload.impl.pool.AsyncLoadThreadPool")) { BTraceUtils.incrementAndGet(rejecctCount); } } @OnTimer(1000) public static void rejectPrintln() { int reject = BTraceUtils.getAndSet(rejecctCount, 0); println(strcat("reject count in 1000 msec: ", str(reject))); AggregationKey key = newAggregationKey("rejectCount"); addToAggregation(histogram, key, reject); addToAggregation(average, key, reject); addToAggregation(max, key, reject); addToAggregation(min, key, reject); addToAggregation(sum, key, reject); addToAggregation(count, key, reject); } @OnEvent public static void onEvent() { BTraceUtils.truncateAggregation(histogram, 10); println("---------------------------------------------"); printAggregation("Count", count); printAggregation("Min", min); printAggregation("Max", max); printAggregation("Average", average); printAggregation("Sum", sum); printAggregation("Histogram", histogram); println("---------------------------------------------"); } }
运行结果:
poolSize : 1 , largestPoolSize = 10 , queueSize = 10 reject count in 1000 msec: 0
说明:
1. poolSize 代表为当前的线程数
2. largestPoolSize 代表为历史最大的线程数
3. queueSize 代表blockqueue的当前堆积的size
4. reject count 代表在1000ms内的被reject的数量
最后
这是我对ThreadPoolExecutor使用过程中的一些经验总结,希望能对大家有所帮助,如有描述不对的地方欢迎拍砖。
发表评论
-
yugong QuickStart
2016-03-05 01:52 0几点说明 a. 数据迁移的方案可参见设计文档,oracl ... -
阿里巴巴开源项目: 阿里巴巴去Oracle数据迁移同步工具
2016-03-05 18:29 6341背景 08年左右,阿里巴巴开始尝试MySQL的相关 ... -
愚公performance
2016-03-02 17:29 0性能测试 全量测试 场景1 (单主键, ... -
yugong AdminGuide
2016-03-02 16:40 0环境要求 操作系统 数据库 迁移方案 部署 ... -
Tddl_hint
2014-01-27 13:52 0背景 工作原理 Hint格式 direct模 ... -
tddl5分库规则
2014-01-26 14:41 0背景 工作原理 构建语法树 元数据 基于 ... -
tddl5优化器
2014-01-22 15:12 0背景 工作原理 构建语法树 元数据 抽象语 ... -
Canal BinlogChange(mariadb5/10)
2014-01-20 17:25 4447背景 先前开源了一个 ... -
asynload quickstart
2013-10-08 22:49 0几点说明: 1. asyncload是做为一个j ... -
网友文档贡献
2013-09-18 15:50 01. Otter源代码解析系列 链接:http://e ... -
Manager配置介绍
2013-09-16 13:00 0通道配置说明 多种同步方式配置 a. 单向同步 ... -
canal&otter FAQ
2013-09-05 17:30 0常见问题 1. canal和 ... -
阿里巴巴开源项目:分布式数据库同步系统otter(解决中美异地机房)
2013-08-22 16:48 40209项目背景 阿里巴巴B2B公司,因为业务的特性 ... -
Otter AdminGuide
2013-08-19 11:06 0几点说明 otter系统自带了manager,所以简化了一 ... -
Otter高可用性
2013-08-17 23:41 0基本需求 网络不可靠,异地机房尤为明显. man ... -
Otter数据一致性
2013-08-17 23:39 0技术选型分析 需要处理一致性的业务场景: 多地修改 ( ... -
Otter扩展性
2013-08-17 22:20 0扩展性定义 按照实现不同,可分为两类: 数据处理自定 ... -
Otter双向回环控制
2013-08-17 21:37 0基本需求 支持mysql/oracle的异构数据库的双 ... -
Otter调度模型
2013-08-17 20:13 0背景 在介绍调度模型之前,首先了解一下otter系统要解 ... -
Otter Manager介绍
2013-08-16 11:16 0背景 otter4.0发布至 ...
相关推荐
ThreadPoolExecutor使用和思考
ThreadPoolExecutor的使用和Android常见的4种线程池使用介绍
在《阿里巴巴java开发手册》中...另外由于前面几种方法内部也是通过ThreadPoolExecutor方式实现,使用ThreadPoolExecutor有助于大家明确线程池的运行规则,创建符合自己的业务场景需要的线程池,避免资源耗尽的风险。
JDK1[1].5中的线程池(ThreadPoolExecutor)使用简介
一个关于java 线程池的例子,也适合android
主要为大家详细介绍了ThreadPoolExecutor线程池的使用方法,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
ThreadPoolExecutor线程池,有详尽介绍,本人进行过测试,可以使用
ThreadPoolExecutor源码解析.pdf
ThreadPoolExecutor源码解析.md
NULL 博文链接:https://bijian1013.iteye.com/blog/2284676
JDK1.5中的线程池(java.util.concurrent.ThreadPoolExecutor)使用
提供工厂方法来创建不同类型的线程池,这篇文章主要介绍了Java ThreadPoolExecutor 线程池的使用介绍,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来...
主要介绍了java ThreadPoolExecutor使用方法简单介绍的相关资料,需要的朋友可以参考下
(转)线程池:java_util_ThreadPoolExecutor 比较详细的介绍了ThreadPoolExecutor用法与属性
在开发过程中,合理地使用线程池能够带来3个好处: 降低资源消耗。 通过重复利用已创建的线程降低线程创建和销毁造成的消耗; 提高响应速度。 当任务到达时,任务可以不需要等到线程创建就能立即执行; 提高...
介绍ThreadPoolExecutor中池和queue配合使用的机制
线程池ThreadPoolExecutor实战及其原理分析(下)线程池ThreadPoolExecutor实战及其原理分析(下)线程池ThreadPoolExecutor实战及其原理分析(下)线程池ThreadPoolExecutor实战及其原理分析(下)线程池ThreadPoolExecutor...
1.资源简介:PyQt5中使用多线程模块QThread解决了PyQt5界面程序执行比较耗时操作时,程序卡顿出现的无响应以及界面输出无法实时显示的问题,采用线程池ThreadPoolExecutor解决了ping多个IP多任务耗时问题。...
线程池执行器 使用多线程ThreadPoolExecutor从Web加载图像