在使用线程处理异步任务时，如果每次执行任务都去创建线程执行完成任务又销毁线程，由于创建线程和销毁线程会需要一些CPU资源，所以我们不得不使用线程池管理线程。
ThreadPollExecutor
ThreadPoolExecutor是创建一个线程池的类，它有四个构造方法，我们看参数最多的那个：
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
解释一些这些参数的作用：
corePoolSize：核心线程数，默认情况下线程池是空的，只有在提交任务时才会创建线程。
在提交任务时会判断，如果当前运行的线程数少于corePoolSize，则会创建新的线程来处理这个任务，如果等于或多于corePoolSize，则不会创建线程，而是将任务加入任务队列等待执行，那如果任务队列满了并且线程池的线程数量小于maximumPoolSize ，那么会创建线程执行任务，如果队列满了并且线程池的线程数量大于等于maximumPoolSize，那么会执行饱和策略。有一个彩蛋，如果你想提前启动线程池中的核心线程，你只要调用prestartAllCoreThread方法，启动所有的核心线程，当然还有prestartCoreThread方法启动一个。maximumPoolSize ：线程池允许创建最大的线程数。如果池中线程等于corePoolSize，并且任务队列也满了，那么会创建新的线程执行任务，前提是池中的线程数小于maximumPoolSize 。
keepAliveTime ：非核心线程闲置的超时时间，超过这个时间那么闲置的线程将被回收，unit是时间单位。
workQueue：任务队列，当前执行任务的线程数大于corePoolSize时，那么后来的任务将被加入workQueue。
*threadFactory：线程工厂，创建线程用的；
- RejectedExecutionHandler ：饱和策略，这是当任务队列满了和当前执行任务的线程数大于等于maximumPoolSize 时即线程池满了，所采取的的策略，以供有四种策略：
1、CallerRunsPolicy：使用调用者当前的线程来处理任务；
2、DiscardPolicy：不执行该任务，并将该任务删除；
3、DiscardOldestPolicy：丢弃队列最近的任务，并执行当前的任务；
4、AbortPolicy：这是默认策略，表示无法执行任务，并抛出RejectedExecutionException异常。
看一下执行逻辑图：
线程池执行过程.png
没有代码，一切都是纸老虎
public class ThreadDemo {
//表示执行饱和策略时退出不在执行任务
static boolean isBreak = false;
public static void main(String[] args) throws ExecutionException, InterruptedException {
//创建线程池
ThreadPoolExecutor executor =
new ThreadPoolExecutor(5, 8, 60, TimeUnit.SECONDS,
new LinkedBlockingDeque<>(10), new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
System.out.println("饱和策略" + r);
isBreak = true;
}
});
while (!isBreak) {
System.out.println("核心线程数: " + executor.getCorePoolSize()
+ ", 线程池最大线程数: " + executor.getMaximumPoolSize()
+ ", 当前线程数: " + executor.getPoolSize()
+ ", 任务队列大小 " + executor.getQueue().size());
TimeUnit.MILLISECONDS.sleep(1000);
executor.execute(() -> {
try {
TimeUnit.SECONDS.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
}
}
最后输出的日志：
核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 0, 任务队列大小 0
核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 1, 任务队列大小 0
核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 2, 任务队列大小 0
核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 3, 任务队列大小 0
核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 3, 任务队列大小 1
核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 3, 任务队列大小 2
核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 3, 任务队列大小 3
核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 4, 任务队列大小 3
核心线程数: 3, 线程池最大线程数: 5, 当前线程数: 5, 任务队列大小 3
饱和策略RejectedExecutionException
在提交任务时会判断，如果当前执行任务的线程数少于corePoolSize，会创建新的线程来处理该任务，如果等于或多于corePoolSize，则不会创建线程，而是将该任务加入任务队列等待执行，那如果任务队列满了并且线程池的线程数量小于maximumPoolSize ，那么会创建非核心线程执行该任务，如果队列满了并且线程池的线程数量大于等于maximumPoolSize，那么会执行饱和策略。
在java中还给我们提供四种线程池，分别是：FixedThreadPoll、CachedThreadPool、SingleThreadPool和ScheduledThreadPool，这四种线程池，无非就是通过ThreadPoolExecutor的不同配置创建出来，具体这四中的实现想了解可以去看一下。
最后附上自己写的SimpleEventBus时封装的的线程池代码：
/**
* 线程切换
*/
public class ScheduleRouterExecutor extends ThreadPoolExecutor {
//主线程
private final Handler mainHandler = new Handler(Looper.getMainLooper());
//CPU数量
private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
//核心线程数量
private static final int INIT_THREAD_COUNT = CPU_COUNT + 1;
//线程池最大县城内容量
private static final int MAX_THREAD_COUNT = INIT_THREAD_COUNT * 2;
//非核心线程闲置存活的时间（秒）
private static final long SURPLUS_THREAD_LIFE = 30L;
//默认线程工厂
private static ThreadFactory DEFAULT_THREAD_FACTORY = new DefaultThreadFactory();
private static RejectedExecutionHandler REJECTED_HANDLER = new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
Log.e("tag", "提交的任务太多了哥们 ，已经执行饱和策略!");
}
};
private static ScheduleRouterExecutor instance;
@NonNull
public static ScheduleRouterExecutor getInstance() {
if (null == instance) {
synchronized (ScheduleRouterExecutor.class) {
if (null == instance) {
instance = new ScheduleRouterExecutor(INIT_THREAD_COUNT, MAX_THREAD_COUNT,
SURPLUS_THREAD_LIFE, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(64),
DEFAULT_THREAD_FACTORY,
REJECTED_HANDLER);
}
}
}
return instance;
}
private ScheduleRouterExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
}
public void executeTask(@NonNull EventTask task) {
if (task.getThreadMode() == ThreadMode.BACKGROUND) {
if (Looper.myLooper() == Looper.getMainLooper()) {
execute(task);
return;
}
task.run();
} else if (task.threadMode == ThreadMode.MAIN) {
if (Looper.myLooper() != Looper.getMainLooper()) {
mainHandler.post(task);
} else {
task.run();
}
}
}
/**
* 线程执行结束，检查是否存在的异常
*
* @param r the runnable that has completed
* @param t the exception that caused termination, or null if
*/
@Override
protected void afterExecute(Runnable r, Throwable t) {
super.afterExecute(r, t);
Log.e("tag", "afterExecute");
if (t == null && r instanceof Future<?>) {
try {
((Future<?>) r).get();
} catch (CancellationException ce) {
t = ce;
} catch (ExecutionException ee) {
t = ee.getCause();
} catch (InterruptedException ie) {
// ignore/reset
Thread.currentThread().interrupt();
}
}
if (t != null) {
Log.e("tag", "Running task appeared exception! Thread [" +
Thread.currentThread().getName() + "], because [" + t.getMessage() + "]\n" +
Arrays.toString(t.getStackTrace()));
}
}
}
线程工厂：
/**
* 线程池工厂类
*/
public class DefaultThreadFactory implements ThreadFactory {
//统计线程池的数量
private static final AtomicInteger poolNumber = new AtomicInteger(1);
//统计当前线程池的线程数量
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final ThreadGroup threadGroup;
private final String namePrefix;
public DefaultThreadFactory() {
SecurityManager securityManager = System.getSecurityManager();
threadGroup = (securityManager != null) ? securityManager.getThreadGroup() :
Thread.currentThread().getThreadGroup();
namePrefix = "Simple EventBus task pool No." + poolNumber.getAndIncrement() + ", thread No.";
}
@Override
public Thread newThread(@NonNull Runnable runnable) {
String threadName = namePrefix + threadNumber.getAndIncrement();
Log.e("tag", "Thread production, name is [" + threadName + "]");
final Thread thread = new Thread(threadGroup, runnable, threadName, 0);
if (thread.isDaemon()) {//设置为非后台线程
thread.setDaemon(false);
}
if (thread.getPriority() != Thread.NORM_PRIORITY) {//优先级为normal
thread.setPriority(Thread.NORM_PRIORITY);
}
// 捕获多线程处理中的异常
thread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
@Override
public void uncaughtException(Thread thread, Throwable ex) {
Log.e("tag", "Running task appeared exception! Thread [" + thread.getName() + "], because [" + ex.getMessage() + "]");
}
});
return thread;
}
}
