Java concurrency线程池之线程池原理(二)_动力节点Java学院整理
|
线程池示例 在分析线程池之前，先看一个简单的线程池示例。
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
public class ThreadPoolDemo1 {
public static void main(String[] args) {
// 创建一个可重用固定线程数的线程池
ExecutorService pool = Executors.newFixedThreadPool(2);
// 创建实现了Runnable接口对象，Thread对象当然也实现了Runnable接口
Thread ta = new MyThread();
Thread tb = new MyThread();
Thread tc = new MyThread();
Thread td = new MyThread();
Thread te = new MyThread();
// 将线程放入池中进行执行
pool.execute(ta);
pool.execute(tb);
pool.execute(tc);
pool.execute(td);
pool.execute(te);
// 关闭线程池
pool.shutdown();
}
}
class MyThread extends Thread {
@Override
public void run() {
System.out.println(Thread.currentThread().getName()+ " is running.");
}
}
运行结果： pool-1-thread-1 is running. pool-1-thread-2 is running. pool-1-thread-1 is running. pool-1-thread-2 is running. pool-1-thread-1 is running. 示例中，包括了线程池的创建，将任务添加到线程池中，关闭线程池这3个主要的步骤。稍后，我们会从这3个方面来分析ThreadPoolExecutor。 线程池源码分析 (一) 创建“线程池” 下面以newFixedThreadPool()介绍线程池的创建过程。 1. newFixedThreadPool() newFixedThreadPool()在Executors.java中定义，源码如下：
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads,nThreads,0L,TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());
}
说明：newFixedThreadPool(int nThreads)的作用是创建一个线程池，线程池的容量是nThreads。 2. ThreadPoolExecutor() ThreadPoolExecutor()在ThreadPoolExecutor.java中定义，源码如下：
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue) {
this(corePoolSize,maximumPoolSize,keepAliveTime,unit,workQueue,Executors.defaultThreadFactory(),defaultHandler);
}
说明：该函数实际上是调用ThreadPoolExecutor的另外一个构造函数。该函数的源码如下：
public ThreadPoolExecutor(int corePoolSize,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;
}
说明：在ThreadPoolExecutor()的构造函数中，进行的是初始化工作。 2.1 ThreadFactory 线程池中的ThreadFactory是一个线程工厂，线程池创建线程都是通过线程工厂对象(threadFactory)来完成的。
public static ThreadFactory defaultThreadFactory() {
return new DefaultThreadFactory();
}
defaultThreadFactory()返回DefaultThreadFactory对象。Executors.java中的DefaultThreadFactory()源码如下：
static class DefaultThreadFactory implements ThreadFactory {
private static final AtomicInteger poolNumber = new AtomicInteger(1);
private final ThreadGroup group;
private final AtomicInteger threadNumber = new AtomicInteger(1);
private final String namePrefix;
DefaultThreadFactory() {
SecurityManager s = System.getSecurityManager();
group = (s != null) ? s.getThreadGroup() :
Thread.currentThread().getThreadGroup();
namePrefix = "pool-" +
poolNumber.getAndIncrement() +
"-thread-";
}
// 提供创建线程的API。
public Thread newThread(Runnable r) {
// 线程对应的任务是Runnable对象r
Thread t = new Thread(group,r,namePrefix + threadNumber.getAndIncrement(),0);
// 设为“非守护线程”
if (t.isDaemon())
t.setDaemon(false);
// 将优先级设为“Thread.NORM_PRIORITY”
if (t.getPriority() != Thread.NORM_PRIORITY)
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}
说明：ThreadFactory的作用就是提供创建线程的功能的线程工厂。 2.2 RejectedExecutionHandler handler是ThreadPoolExecutor中拒绝策略的处理句柄。所谓拒绝策略，是指将任务添加到线程池中时，线程池拒绝该任务所采取的相应策略。 private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
public static class AbortPolicy implements RejectedExecutionHandler {
public AbortPolicy() { }
// 抛出异常
public void rejectedExecution(Runnable r,ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
(二) 添加任务到“线程池” 1. execute() execute()定义在ThreadPoolExecutor.java中，源码如下：
public void execute(Runnable command) {
// 如果任务为null，则抛出异常。
if (command == null)
throw new NullPointerException();
// 获取ctl对应的int值。该int值保存了"线程池中任务的数量"和"线程池状态"信息
int c = ctl.get();
// 当线程池中的任务数量 < "核心池大小"时，即线程池中少于corePoolSize个任务。
// 则通过addWorker(command,true)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command,true))
return;
c = ctl.get();
}
// 当线程池中的任务数量 >= "核心池大小"时，
// 而且，"线程池处于允许状态"时，则尝试将任务添加到阻塞队列中。
if (isRunning(c) && workQueue.offer(command)) {
// 再次确认“线程池状态”，若线程池异常终止了，则删除任务；然后通过reject()执行相应的拒绝策略的内容。
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
// 否则，如果"线程池中任务数量"为0，则通过addWorker(null,false)尝试新建一个线程，新建线程对应的任务为null。
else if (workerCountOf(recheck) == 0)
addWorker(null,false);
}
// 通过addWorker(command,false)新建一个线程，并将任务(command)添加到该线程中；然后，启动该线程从而执行任务。
// 如果addWorker(command,false)执行失败，则通过reject()执行相应的拒绝策略的内容。
else if (!addWorker(command,false))
reject(command);
}
说明：execute()的作用是将任务添加到线程池中执行。它会分为3种情况进行处理： 2. addWorker() addWorker()的源码如下： （编辑：安卓应用网） 【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容! |