Java 自定义线程池和线程总数控制操作
作者：JAVA前线
1 概述
池化是常见的思想，线程池是非常典型的池化的实现，《Java并发编程实战》也大篇幅去讲解了Java中的线程池。本文实现一个简单的线程池。
2 核心类
【1】接口定义
public interface IThreadPool<Job extends Runnable> {
/**
* 关闭线程池
*/
public void shutAlldown();
/**
* 执行任务
*
* @param job 任务
*/
public void execute(Job job);
/**
* 添加工作者
*
* @param addNum 添加数
*/
public void addWorkers(int addNum);
/**
* 减少工作者
*
* @param reduceNum 减少数目
*/
public void reduceWorkers(int reduceNum);
}
【2】实现类
线程池的核心是维护了1个任务列表和1个工作者列表。
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
public class XYThreadPool<Job extends Runnable> implements IThreadPool<Job> {
// 默认线程数
private static int DEAFAULT_SIZE = 5;
// 最大线程数
private static int MAX_SIZE = 10;
// 任务列表
private LinkedList<Job> tasks = new LinkedList<Job>();
// 工作线程列表
private List<Worker> workers = Collections
.synchronizedList(new ArrayList<Worker>());
/**
* 默认构造函数
*/
public XYThreadPool() {
initWokers(DEAFAULT_SIZE);
}
/**
* 执行线程数
*
* @param threadNums 线程数
*/
public XYThreadPool(int workerNum) {
workerNum = workerNum <= 0 ? DEAFAULT_SIZE
: workerNum > MAX_SIZE ? MAX_SIZE : workerNum;
initWokers(workerNum);
}
/**
* 初始化线程池
*
* @param threadNums 线程数
*/
public void initWokers(int threadNums) {
for (int i = 0; i < threadNums; i++) {
Worker worker = new Worker();
worker.start();
workers.add(worker);
}
// 添加关闭钩子
Runtime.getRuntime().addShutdownHook(new Thread() {
public void run() {
shutAlldown();
}
});
}
@Override
public void shutAlldown() {
for (Worker worker : workers) {
worker.shutdown();
}
}
@Override
public void execute(Job job) {
synchronized (tasks) {
// 提交任务就是将任务对象加入任务队列，等待工作线程去处理
tasks.addLast(job);
tasks.notifyAll();
}
}
@Override
public void addWorkers(int addNum) {
// 新线程数必须大于零，并且线程总数不能大于最大线程数
if ((workers.size() + addNum) <= MAX_SIZE && addNum > 0) {
initWokers(addNum);
} else {
System.out.println("addNum too large");
}
}
@Override
public void reduceWorkers(int reduceNum) {
if ((workers.size() - reduceNum <= 0))
System.out.println("thread num too small");
else {
// 暂停指定数量的工作者
int count = 0;
while (count != reduceNum) {
for (Worker w : workers) {
w.shutdown();
count++;
}
}
}
}
/**
* 工作线程
*/
class Worker extends Thread {
private volatile boolean flag = true;
@Override
public void run() {
while (flag) {
Job job = null;
// 加锁（若只有一个woker可不必加锁，那就是所谓的单线程的线程池，线程安全）
synchronized (tasks) {
// 任务队列为空
while (tasks.isEmpty()) {
try {
// 阻塞，放弃对象锁，等待被notify唤醒
tasks.wait();
System.out.println("block when tasks is empty");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 不为空取出任务
job = tasks.removeFirst();
System.out.println("get job:" + job + ",do biz");
job.run();
}
}
}
public void shutdown() {
flag = false;
}
}
}
(1) 当调用wait()方法时线程会放弃对象锁，进入等待此对象的等待锁定池，只有针对此对象调用notify()方法后本线程才进入对象锁定池准备
(2) Object的方法：void notify(): 唤醒一个正在等待该对象的线程。void notifyAll(): 唤醒所有正在等待该对象的线程。
notifyAll使所有原来在该对象上等待被notify的线程统统退出wait状态，变成等待该对象上的锁，一旦该对象被解锁，它们会去竞争。
notify只是选择一个wait状态线程进行通知，并使它获得该对象上的锁，但不惊动其它同样在等待被该对象notify的线程们，当第一个线程运行完毕以后释放对象上的锁，此时如果该对象没有再次使用notify语句，即便该对象已经空闲，其他wait状态等待的线程由于没有得到该对象的通知，继续处在wait状态，直到这个对象发出一个notify或notifyAll，它们等待的是被notify或notifyAll，而不是锁。
3 无需控制线程总数
每调用一次就会创建一个拥有10个线程工作者的线程池。
public class TestService1 {
public static void main(String[] args) {
// 启动10个线程
XYThreadPool<Runnable> pool = new XYThreadPool<Runnable>(10);
pool.execute(new Runnable() {
@Override
public void run() {
System.out.println("====1 test====");
}
});
}
}
public class TestService2 {
public static void main(String[] args) {
// 启动10个线程
XYThreadPool<Runnable> pool = new XYThreadPool<Runnable>(10);
pool.execute(new Runnable() {
@Override
public void run() {
System.out.println("====2 test====");
}
});
}
}
4 控制线程总数
在项目中所有的线程调用，一般都共用1个固定工作者数大小的线程池。
import javax.annotation.PostConstruct;
import org.springframework.stereotype.Component;
import com.xy.pool.XYThreadPool;
/**
* 统一线程池管理类
*/
@Component
public class XYThreadManager {
private XYThreadPool<Runnable> executorPool;
@PostConstruct
public void init() {
executorPool = new XYThreadPool<Runnable>(10);
}
public XYThreadPool<Runnable> getExecutorPool() {
return executorPool;
}
}
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service("testService3")
public class TestService3 {
@Autowired
private XYThreadManager threadManager;
public void test() {
threadManager.getExecutorPool().execute(new Runnable() {
@Override
public void run() {
System.out.println("====3 test====");
}
});
}
}
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service("testService4")
public class TestService4 {
@Autowired
private XYThreadManager threadManager;
public void test() {
threadManager.getExecutorPool().execute(new Runnable() {
@Override
public void run() {
System.out.println("====4 test====");
}
});
}
}
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class TestMain {
@SuppressWarnings("resource")
public static void main(String[] args) {
ApplicationContext atc = new ClassPathXmlApplicationContext("applicationContext.xml");
TestService3 t3 = (TestService3) atc.getBean("testService3");
t3.test();
TestService4 t4 = (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，当然这都不是绝对的，需要根据实际情况以及实际业务来调整：final int poolSize = (int)(cpuCore/(1-0.9))
针对于阻塞系数，《Programming Concurrency on the JVM Mastering》即《Java 虚拟机并发编程》中有提到一句话：
对于阻塞系数，我们可以先试着猜测，抑或采用一些细嫩分析工具或java.lang.management API 来确定线程花在系统/IO操作上的时间与CPU密集任务所耗的时间比值。
线程池相关参数配置
说到这一点，我们只需要谨记一点，一定不要选择没有上限限制的配置项。
这也是为什么不建议使用Executors 中创建线程的方法。
比如，Executors.newCachedThreadPool的设置与无界队列的设置因为某些不可预期的情况，线程池会出现系统异常，导致线程暴增的情况或者任务队列不断膨胀，内存耗尽导致系统崩溃和异常。 我们推荐使用自定义线程池来避免该问题，这也是在使用线程池规范的首要原则！ 小心无大错，千万别过度自信！
可以看下Executors中四个创建线程池的方法：
//使用无界队列
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
//线程池数量是无限的
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
其他的就不再列举了，大家可以自行查阅源码。
第二，合理设置线程数量、和线程空闲回收时间，根据具体的任务执行周期和时间去设定，避免频繁的回收和创建，虽然我们使用线程池的目的是为了提升系统性能和吞吐量，但是也要考虑下系统的稳定性，不然出现不可预期问题会很麻烦！
第三，根据实际场景，选择适用于自己的拒绝策略。进行补偿，不要乱用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方法，手工去调用线程池的关闭方法！
线程池使用实例
线程池核心代码：
public class AsyncProcessQueue {
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
/**
* Task 包装类<br>
* 此类型的意义是记录可能会被 Executor 吃掉的异常<br>
*/
public static class TaskWrapper implements Runnable {
private static final Logger _LOGGER = LoggerFactory.getLogger(TaskWrapper.class);
private final Runnable gift;
public TaskWrapper(final Runnable target) {
this.gift = target;
}
@Override
public void run() {
// 捕获异常，避免在 Executor 里面被吞掉了
if (gift != null) {
try {
gift.run();
} catch (Exception e) {
_LOGGER.error("Wrapped target execute exception.", e);
}
}
}
}
// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
/**
* 执行指定的任务
*
* @param task
* @return
*/
public static boolean execute(final Runnable task) {
return AsyncProcessor.executeTask(new TaskWrapper(task));
}
}
public class AsyncProcessor {
static final Logger LOGGER = LoggerFactory.getLogger(AsyncProcessor.class);
/**
* 默认最大并发数<br>
*/
private static final int DEFAULT_MAX_CONCURRENT = Runtime.getRuntime().availableProcessors() * 2;
/**
* 线程池名称格式
*/
private static final String THREAD_POOL_NAME = "ExternalConvertProcessPool-%d";
/**
* 线程工厂名称
*/
private static final ThreadFactory FACTORY = new BasicThreadFactory.Builder().namingPattern(THREAD_POOL_NAME)
.daemon(true).build();
/**
* 默认队列大小
*/
private static final int DEFAULT_SIZE = 500;
/**
* 默认线程存活时间
*/
private static final long DEFAULT_KEEP_ALIVE = 60L;
/**NewEntryServiceImpl.java:689
* Executor
*/
private static ExecutorService executor;
/**
* 执行队列
*/
private static BlockingQueue<Runnable> executeQueue = new ArrayBlockingQueue<>(DEFAULT_SIZE);
static {
// 创建 Executor
// 此处默认最大值改为处理器数量的 4 倍
try {
executor = new ThreadPoolExecutor(DEFAULT_MAX_CONCURRENT, DEFAULT_MAX_CONCURRENT * 4, DEFAULT_KEEP_ALIVE,
TimeUnit.SECONDS, executeQueue, FACTORY);
// 关闭事件的挂钩
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
@Override
public void run() {
AsyncProcessor.LOGGER.info("AsyncProcessor shutting down.");
executor.shutdown();
try {
// 等待1秒执行关闭
if (!executor.awaitTermination(1, TimeUnit.SECONDS)) {
AsyncProcessor.LOGGER.error("AsyncProcessor shutdown immediately due to wait timeout.");
executor.shutdownNow();
}
} catch (InterruptedException e) {
AsyncProcessor.LOGGER.error("AsyncProcessor shutdown interrupted.");
executor.shutdownNow();
}
AsyncProcessor.LOGGER.info("AsyncProcessor shutdown complete.");
}
}));
} catch (Exception e) {
LOGGER.error("AsyncProcessor init error.", e);
throw new ExceptionInInitializerError(e);
}
}
/**
* 此类型无法实例化
*/
private AsyncProcessor() {
}
/**
* 执行任务，不管是否成功<br>
* 其实也就是包装以后的 {@link Executer} 方法
*
* @param task
* @return
*/
public static boolean executeTask(Runnable task) {
try {
executor.execute(task);
} catch (RejectedExecutionException e) {
LOGGER.error("Task executing was rejected.", e);
return false;
}
return true;
}
/**
* 提交任务，并可以在稍后获取其执行情况<br>
* 当提交失败时，会抛出 {@link }
*
* @param task
* @return
*/
public static <T> Future<T> submitTask(Callable<T> task) {
try {
return executor.submit(task);
} catch (RejectedExecutionException e) {
LOGGER.error("Task executing was rejected.", e);
throw new UnsupportedOperationException("Unable to submit the task, rejected.", e);
}
}
}
使用方式：
AsyncProcessQueue.execute(new Runnable() {
@Override
public void run() {
//do something
}
});
可以根据自己的使用场景灵活变更，我这里并没有用到beforeExecute和afterExecute以及拒绝策略。
以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。如有错误或未考虑完全的地方，望不吝赐教。