Java 自定义线程池和线程总数控制操作

2023-07-05 01:16:03 206

1概述

池化是常见的思想，线程池是非常典型的池化的实现，《Java并发编程实战》也大篇幅去讲解了Java中的线程池。本文实现一个简单的线程池。

2核心类

【1】接口定义

publicinterfaceIThreadPool{
/**
*关闭线程池
*/
publicvoidshutAlldown();

/**
*执行任务
*
*@paramjob任务
*/
publicvoidexecute(Jobjob);

/**
*添加工作者
*
*@paramaddNum添加数
*/
publicvoidaddWorkers(intaddNum);

/**
*减少工作者
*
*@paramreduceNum减少数目
*/
publicvoidreduceWorkers(intreduceNum);
}

【2】实现类

线程池的核心是维护了1个任务列表和1个工作者列表。

importjava.util.ArrayList;
importjava.util.Collections;
importjava.util.LinkedList;
importjava.util.List;
publicclassXYThreadPoolimplementsIThreadPool{
//默认线程数
privatestaticintDEAFAULT_SIZE=5;
//最大线程数
privatestaticintMAX_SIZE=10;
//任务列表
privateLinkedListtasks=newLinkedList();
//工作线程列表
privateListworkers=Collections
.synchronizedList(newArrayList());
/**
*默认构造函数
*/
publicXYThreadPool(){
initWokers(DEAFAULT_SIZE);
}
/**
*执行线程数
*
*@paramthreadNums线程数
*/
publicXYThreadPool(intworkerNum){
workerNum=workerNum<=0?DEAFAULT_SIZE
:workerNum>MAX_SIZE?MAX_SIZE:workerNum;
initWokers(workerNum);
}
/**
*初始化线程池
*
*@paramthreadNums线程数
*/
publicvoidinitWokers(intthreadNums){
for(inti=0;i0){
initWokers(addNum);
}else{
System.out.println("addNumtoolarge");
}
}
@Override
publicvoidreduceWorkers(intreduceNum){
if((workers.size()-reduceNum<=0))
System.out.println("threadnumtoosmall");
else{
//暂停指定数量的工作者
intcount=0;
while(count!=reduceNum){
for(Workerw:workers){
w.shutdown();
count++;
}
}
}
}
/**
*工作线程
*/
classWorkerextendsThread{
privatevolatilebooleanflag=true;
@Override
publicvoidrun(){
while(flag){
Jobjob=null;
//加锁（若只有一个woker可不必加锁，那就是所谓的单线程的线程池，线程安全）
synchronized(tasks){
//任务队列为空
while(tasks.isEmpty()){
try{
//阻塞，放弃对象锁，等待被notify唤醒
tasks.wait();
System.out.println("blockwhentasksisempty");
}catch(InterruptedExceptione){
e.printStackTrace();
}
}
//不为空取出任务
job=tasks.removeFirst();
System.out.println("getjob:"+job+",dobiz");
job.run();
}
}
}
publicvoidshutdown(){
flag=false;
}
}
}

(1)当调用wait()方法时线程会放弃对象锁，进入等待此对象的等待锁定池，只有针对此对象调用notify()方法后本线程才进入对象锁定池准备

(2)Object的方法：voidnotify():唤醒一个正在等待该对象的线程。voidnotifyAll():唤醒所有正在等待该对象的线程。

notifyAll使所有原来在该对象上等待被notify的线程统统退出wait状态，变成等待该对象上的锁，一旦该对象被解锁，它们会去竞争。

notify只是选择一个wait状态线程进行通知，并使它获得该对象上的锁，但不惊动其它同样在等待被该对象notify的线程们，当第一个线程运行完毕以后释放对象上的锁，此时如果该对象没有再次使用notify语句，即便该对象已经空闲，其他wait状态等待的线程由于没有得到该对象的通知，继续处在wait状态，直到这个对象发出一个notify或notifyAll，它们等待的是被notify或notifyAll，而不是锁。

3无需控制线程总数

每调用一次就会创建一个拥有10个线程工作者的线程池。

publicclassTestService1{
publicstaticvoidmain(String[]args){
//启动10个线程
XYThreadPoolpool=newXYThreadPool(10);
pool.execute(newRunnable(){
@Override
publicvoidrun(){
System.out.println("====1test====");
}
});
}
}
publicclassTestService2{
publicstaticvoidmain(String[]args){
//启动10个线程
XYThreadPoolpool=newXYThreadPool(10);
pool.execute(newRunnable(){
@Override
publicvoidrun(){
System.out.println("====2test====");
}
});
}
}

4控制线程总数

在项目中所有的线程调用，一般都共用1个固定工作者数大小的线程池。

importjavax.annotation.PostConstruct;
importorg.springframework.stereotype.Component;
importcom.xy.pool.XYThreadPool;
/**
*统一线程池管理类
*/
@Component
publicclassXYThreadManager{
privateXYThreadPoolexecutorPool;
@PostConstruct
publicvoidinit(){
executorPool=newXYThreadPool(10);
}
publicXYThreadPoolgetExecutorPool(){
returnexecutorPool;
}
}
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.stereotype.Service;
@Service("testService3")
publicclassTestService3{
@Autowired
privateXYThreadManagerthreadManager;
publicvoidtest(){
threadManager.getExecutorPool().execute(newRunnable(){
@Override
publicvoidrun(){
System.out.println("====3test====");
}
});
}
}
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.stereotype.Service;
@Service("testService4")
publicclassTestService4{
@Autowired
privateXYThreadManagerthreadManager;
publicvoidtest(){
threadManager.getExecutorPool().execute(newRunnable(){
@Override
publicvoidrun(){
System.out.println("====4test====");
}
});
}
}
importorg.springframework.context.ApplicationContext;
importorg.springframework.context.support.ClassPathXmlApplicationContext;
publicclassTestMain{
@SuppressWarnings("resource")
publicstaticvoidmain(String[]args){
ApplicationContextatc=newClassPathXmlApplicationContext("applicationContext.xml");
TestService3t3=(TestService3)atc.getBean("testService3");
t3.test();
TestService4t4=(TestService4)atc.getBean("testService4");
t4.test();
}
}

补充：论如何优雅的自定义ThreadPoolExecutor线程池

前言

线程池想必大家也都用过，JDK的Executors也自带一些线程池。但是不知道大家有没有想过，如何才是最优雅的方式去使用过线程池吗？生产环境要怎么去配置自己的线程池才是合理的呢？

今天周末，刚好有时间来总结一下自己所认为的'优雅',如有问题欢迎大家指正。

线程池使用规则

要使用好线程池，那么一定要遵循几个规则：

线程个数大小的设置

线程池相关参数配置

利用Hook嵌入你的行为

线程池的关闭

线程池配置相关

线程池大小的设置

这其实是一个面试的考点，很多面试官会问你线程池coreSize的大小来考察你对于线程池的理解。

首先针对于这个问题，我们必须要明确我们的需求是计算密集型还是IO密集型，只有了解了这一点，我们才能更好的去设置线程池的数量进行限制。

1、计算密集型：

顾名思义就是应用需要非常多的CPU计算资源，在多核CPU时代，我们要让每一个CPU核心都参与计算，将CPU的性能充分利用起来，这样才算是没有浪费服务器配置，如果在非常好的服务器配置上还运行着单线程程序那将是多么重大的浪费。对于计算密集型的应用，完全是靠CPU的核数来工作，所以为了让它的优势完全发挥出来，避免过多的线程上下文切换，比较理想方案是：

线程数=CPU核数+1，也可以设置成CPU核数*2，但还要看JDK的版本以及CPU配置(服务器的CPU有超线程)。

一般设置CPU*2即可。

2、IO密集型

我们现在做的开发大部分都是WEB应用，涉及到大量的网络传输，不仅如此，与数据库，与缓存间的交互也涉及到IO，一旦发生IO，线程就会处于等待状态，当IO结束，数据准备好后，线程才会继续执行。因此从这里可以发现，对于IO密集型的应用，我们可以多设置一些线程池中线程的数量，这样就能让在等待IO的这段时间内，线程可以去做其它事，提高并发处理效率。那么这个线程池的数据量是不是可以随便设置呢？当然不是的，请一定要记得，线程上下文切换是有代价的。目前总结了一套公式，对于IO密集型应用：

线程数=CPU核心数/(1-阻塞系数)这个阻塞系数一般为0.8~0.9之间，也可以取0.8或者0.9。

套用公式，对于双核CPU来说，它比较理想的线程数就是20，当然这都不是绝对的，需要根据实际情况以及实际业务来调整：finalintpoolSize=(int)(cpuCore/(1-0.9))

针对于阻塞系数，《ProgrammingConcurrencyontheJVMMastering》即《Java虚拟机并发编程》中有提到一句话：

对于阻塞系数，我们可以先试着猜测，抑或采用一些细嫩分析工具或java.lang.managementAPI来确定线程花在系统/IO操作上的时间与CPU密集任务所耗的时间比值。

线程池相关参数配置

说到这一点，我们只需要谨记一点，一定不要选择没有上限限制的配置项。

这也是为什么不建议使用Executors中创建线程的方法。

比如，Executors.newCachedThreadPool的设置与无界队列的设置因为某些不可预期的情况，线程池会出现系统异常，导致线程暴增的情况或者任务队列不断膨胀，内存耗尽导致系统崩溃和异常。我们推荐使用自定义线程池来避免该问题，这也是在使用线程池规范的首要原则！小心无大错，千万别过度自信！

可以看下Executors中四个创建线程池的方法：

//使用无界队列
publicstaticExecutorServicenewFixedThreadPool(intnThreads){
returnnewThreadPoolExecutor(nThreads,nThreads,
0L,TimeUnit.MILLISECONDS,
newLinkedBlockingQueue());
}

//线程池数量是无限的
publicstaticExecutorServicenewCachedThreadPool(){
returnnewThreadPoolExecutor(0,Integer.MAX_VALUE,
60L,TimeUnit.SECONDS,
newSynchronousQueue());
}

其他的就不再列举了，大家可以自行查阅源码。

第二，合理设置线程数量、和线程空闲回收时间，根据具体的任务执行周期和时间去设定，避免频繁的回收和创建，虽然我们使用线程池的目的是为了提升系统性能和吞吐量，但是也要考虑下系统的稳定性，不然出现不可预期问题会很麻烦！

第三，根据实际场景，选择适用于自己的拒绝策略。进行补偿，不要乱用JDK支持的自动补偿机制！尽量采用自定义的拒绝策略去进行兜底！

第四，线程池拒绝策略，自定义拒绝策略可以实现RejectedExecutionHandler接口。

JDK自带的拒绝策略如下：

AbortPolicy：直接抛出异常阻止系统正常工作。

CallerRunsPolicy：只要线程池未关闭，该策略直接在调用者线程中，运行当前被丢弃的任务。

DiscardOldestPolicy：丢弃最老的一个请求，尝试再次提交当前任务。

DiscardPolicy：丢弃无法处理的任务，不给予任何处理。

利用Hook

利用Hook，留下线程池执行轨迹：

ThreadPoolExecutor提供了protected类型可以被覆盖的钩子方法，允许用户在任务执行之前会执行之后做一些事情。我们可以通过它来实现比如初始化ThreadLocal、收集统计信息、如记录日志等操作。这类Hook如beforeExecute和afterExecute。另外还有一个Hook可以用来在任务被执行完的时候让用户插入逻辑，如rerminated。

如果hook方法执行失败，则内部的工作线程的执行将会失败或被中断。

我们可以使用beforeExecute和afterExecute来记录线程之前前和后的一些运行情况，也可以直接把运行完成后的状态记录到ELK等日志系统。

关闭线程池

内容当线程池不在被引用并且工作线程数为0的时候，线程池将被终止。我们也可以调用shutdown来手动终止线程池。如果我们忘记调用shutdown，为了让线程资源被释放，我们还可以使用keepAliveTime和allowCoreThreadTimeOut来达到目的!

当然，稳妥的方式是使用虚拟机Runtime.getRuntime().addShutdownHook方法，手工去调用线程池的关闭方法！

线程池使用实例

线程池核心代码：

publicclassAsyncProcessQueue{
//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
/**
*Task包装类

*此类型的意义是记录可能会被Executor吃掉的异常

*/
publicstaticclassTaskWrapperimplementsRunnable{
privatestaticfinalLogger_LOGGER=LoggerFactory.getLogger(TaskWrapper.class);
privatefinalRunnablegift;
publicTaskWrapper(finalRunnabletarget){
this.gift=target;
}
@Override
publicvoidrun(){

//捕获异常，避免在Executor里面被吞掉了
if(gift!=null){

try{
gift.run();
}catch(Exceptione){
_LOGGER.error("Wrappedtargetexecuteexception.",e);
}
}
}
}
//~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

/**
*执行指定的任务
*
*@paramtask
*@return
*/
publicstaticbooleanexecute(finalRunnabletask){
returnAsyncProcessor.executeTask(newTaskWrapper(task));
}
}
publicclassAsyncProcessor{
staticfinalLoggerLOGGER=LoggerFactory.getLogger(AsyncProcessor.class);

/**
*默认最大并发数

*/
privatestaticfinalintDEFAULT_MAX_CONCURRENT=Runtime.getRuntime().availableProcessors()*2;

/**
*线程池名称格式
*/
privatestaticfinalStringTHREAD_POOL_NAME="ExternalConvertProcessPool-%d";

/**
*线程工厂名称
*/
privatestaticfinalThreadFactoryFACTORY=newBasicThreadFactory.Builder().namingPattern(THREAD_POOL_NAME)
.daemon(true).build();

/**
*默认队列大小
*/
privatestaticfinalintDEFAULT_SIZE=500;

/**
*默认线程存活时间
*/
privatestaticfinallongDEFAULT_KEEP_ALIVE=60L;

/**NewEntryServiceImpl.java:689
*Executor
*/
privatestaticExecutorServiceexecutor;

/**
*执行队列
*/
privatestaticBlockingQueueexecuteQueue=newArrayBlockingQueue<>(DEFAULT_SIZE);
static{
//创建Executor
//此处默认最大值改为处理器数量的4倍
try{
executor=newThreadPoolExecutor(DEFAULT_MAX_CONCURRENT,DEFAULT_MAX_CONCURRENT*4,DEFAULT_KEEP_ALIVE,
TimeUnit.SECONDS,executeQueue,FACTORY);
//关闭事件的挂钩
Runtime.getRuntime().addShutdownHook(newThread(newRunnable(){
@Override
publicvoidrun(){
AsyncProcessor.LOGGER.info("AsyncProcessorshuttingdown.");
executor.shutdown();
try{
//等待1秒执行关闭
if(!executor.awaitTermination(1,TimeUnit.SECONDS)){
AsyncProcessor.LOGGER.error("AsyncProcessorshutdownimmediatelyduetowaittimeout.");
executor.shutdownNow();
}
}catch(InterruptedExceptione){
AsyncProcessor.LOGGER.error("AsyncProcessorshutdowninterrupted.");
executor.shutdownNow();
}
AsyncProcessor.LOGGER.info("AsyncProcessorshutdowncomplete.");
}
}));
}catch(Exceptione){
LOGGER.error("AsyncProcessoriniterror.",e);
thrownewExceptionInInitializerError(e);
}
}
/**
*此类型无法实例化
*/
privateAsyncProcessor(){
}
/**
*执行任务，不管是否成功

*其实也就是包装以后的{@linkExecuter}方法
*
*@paramtask
*@return
*/
publicstaticbooleanexecuteTask(Runnabletask){
try{
executor.execute(task);
}catch(RejectedExecutionExceptione){
LOGGER.error("Taskexecutingwasrejected.",e);
returnfalse;
}
returntrue;
}
/**
*提交任务，并可以在稍后获取其执行情况

*当提交失败时，会抛出{@link}
*
*@paramtask
*@return
*/
publicstaticFuturesubmitTask(Callabletask){
try{
returnexecutor.submit(task);
}catch(RejectedExecutionExceptione){
LOGGER.error("Taskexecutingwasrejected.",e);
thrownewUnsupportedOperationException("Unabletosubmitthetask,rejected.",e);
}
}
}

使用方式：

AsyncProcessQueue.execute(newRunnable(){
@Override
publicvoidrun(){
//dosomething
}
});

可以根据自己的使用场景灵活变更，我这里并没有用到beforeExecute和afterExecute以及拒绝策略。

以上为个人经验，希望能给大家一个参考，也希望大家多多支持毛票票。如有错误或未考虑完全的地方，望不吝赐教。

声明：本文内容来源于网络，版权归原作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎发送邮件至：czq8825#qq.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。

Java 自定义线程池和线程总数控制操作

1概述

2核心类

【1】接口定义

【2】实现类

3无需控制线程总数

4控制线程总数

前言

线程池使用规则

线程个数大小的设置

线程池配置相关

线程池相关参数配置

利用Hook

关闭线程池

线程池使用实例

热门推荐

随机推荐