首页 > 代码库 > Java 线程池分析
Java 线程池分析
在项目中经常会用到java线程池,但是别人问起线程池的原理,线程池的策略怎么实现的? 答得不太好,所以按照源码分析一番,首先看下最常用的线程池代码:
public class ThreadPoolTest { private static Executor executor= Executors.newFixedThreadPool(10); //一般通过fixThreadPool起线程池 public static void main(String[] args){ for(int i=0;i<50;i++){ executor.execute(new Task()); //通过线程池执行task } } static class Task implements Runnable{ //Task任务,打印当前执行该任务的线程名称 @Override public void run() { System.out.println(Thread.currentThread().getName()); } } }
这个输出结果为,可以看到,没有线程池的超过10的,所以线程池执行只用了10个线程:
pool-1-thread-1
pool-1-thread-2
pool-1-thread-3
pool-1-thread-4
pool-1-thread-5
pool-1-thread-6
pool-1-thread-7
pool-1-thread-8
pool-1-thread-9
pool-1-thread-10
pool-1-thread-10
pool-1-thread-2
pool-1-thread-3
pool-1-thread-5
pool-1-thread-2
pool-1-thread-4
pool-1-thread-2
pool-1-thread-3
pool-1-thread-2
接下来分析一下源码,在newFixedThreadPool(10)的时候,源码是这样的,返回一个ThreadPoolExecutor
public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); }
再看下ThreadPoolExecutor的构造方法,如下,参数说明我标在注释中了,默认参数见上面传的。所以在newFixedThreadPool(10)时,其实只传了一个参数,其他的都为默认。
public ThreadPoolExecutor(int corePoolSize, //线程池核心池大小,当提交一个任务时,线程池会新创建一个线程池来执行任务,直到线程池大小等于该值。等于后继续提交的任务被保存在阻塞队列,等待执行。 int maximumPoolSize, //最大线程池大小,当阻塞队列满后,如果继续提交任务,则创建新的线程执行任务,前提是当前线程数小于maximumPoolSize; long keepAliveTime, //线程池keepalive时间,即空闲线程池的存活时间。 TimeUnit unit, //keepAliveTime的单位 BlockingQueue<Runnable> workQueue) //阻塞队列,比如实现Runnable接口{ this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); }
说明:在JDK中提供了如下阻塞队列:
2、LinkedBlockingQuene:基于链表结构的阻塞队列,按FIFO排序任务,吞吐量通常要高于ArrayBlockingQuene;
3、SynchronousQuene:一个不存储元素的阻塞队列,每个插入操作必须等到另一个线程调用移除操作,否则插入操作一直处于阻塞状态,吞吐量通常要高于LinkedBlockingQuene;
4、priorityBlockingQuene:具有优先级的无界阻塞队列;
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() { //设置线程名,group等 SecurityManager s = System.getSecurityManager(); group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-"; }
private static final RejectedExecutionHandler defaultHandler = //默认饱和策略 new AbortPolicy();
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; }
public class ThreadPoolTest { private static Executor executor1= new ThreadPoolExecutor(10,20,0, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<Runnable>(3)); public static void main(String[] args){ for(int i=0;i<50;i++){ executor.execute(new Task()); } } static class Task implements Runnable{ @Override public void run() { System.out.println(Thread.currentThread().getName()); } } }
执行结果如下,确实看到了20个线程的名称:
pool-1-thread-1
pool-1-thread-3
pool-1-thread-2
pool-1-thread-5
pool-1-thread-4
pool-1-thread-6
pool-1-thread-7
pool-1-thread-8
pool-1-thread-9
pool-1-thread-10
pool-1-thread-1
pool-1-thread-1
pool-1-thread-1
pool-1-thread-1
pool-1-thread-11
pool-1-thread-12
pool-1-thread-12
pool-1-thread-12
pool-1-thread-12
pool-1-thread-12
pool-1-thread-13
pool-1-thread-14
pool-1-thread-14
pool-1-thread-14
pool-1-thread-14
pool-1-thread-15
pool-1-thread-16
pool-1-thread-16
pool-1-thread-16
pool-1-thread-16
pool-1-thread-17
pool-1-thread-17
pool-1-thread-17
pool-1-thread-17
pool-1-thread-18
pool-1-thread-18
pool-1-thread-18
pool-1-thread-18
pool-1-thread-19
pool-1-thread-19
pool-1-thread-19
pool-1-thread-19
pool-1-thread-20
pool-1-thread-20
pool-1-thread-20
pool-1-thread-20
pool-1-thread-1
pool-1-thread-3
pool-1-thread-2
pool-1-thread-4
pool-1-thread-5
pool-1-thread-6
pool-1-thread-7
pool-1-thread-8
pool-1-thread-9
pool-1-thread-9
pool-1-thread-9
pool-1-thread-9
pool-1-thread-10
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task ThreadPoolTest$Task@4dc63996 rejected from java.util.concurrent.ThreadPoolExecutor@d716361[Running, pool size = 10, active threads = 0, queued tasks = 0, completed tasks = 13]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
线程池的源码及调度流程分析完成了,接下来看下线程池的executor怎么去执行的
先看下线程的状态
1、RUNNING:
-1 << COUNT_BITS
,即高3位为111,该状态的线程池会接收新任务,并处理阻塞队列中的任务;2、SHUTDOWN:
0 << COUNT_BITS
,即高3位为000,该状态的线程池不会接收新任务,但会处理阻塞队列中的任务;3、STOP :
1 << COUNT_BITS
,即高3位为001,该状态的线程不会接收新任务,也不会处理阻塞队列中的任务,而且会中断正在运行的任务;4、TIDYING :
2 << COUNT_BITS
,即高3位为010;5、TERMINATED:
3 << COUNT_BITS
,即高3位为011;public void execute(Runnable command) { if (command == null) throw new NullPointerException(); /* * Proceed in 3 steps: * * 1. If fewer than corePoolSize threads are running, try to * start a new thread with the given command as its first * task. The call to addWorker atomically checks runState and * workerCount, and so prevents false alarms that would add * threads when it shouldn‘t, by returning false. * * 2. If a task can be successfully queued, then we still need * to double-check whether we should have added a thread * (because existing ones died since last checking) or that * the pool shut down since entry into this method. So we * recheck state and if necessary roll back the enqueuing if * stopped, or start a new thread if there are none. * * 3. If we cannot queue task, then we try to add a new * thread. If it fails, we know we are shut down or saturated * and so reject the task. */ int c = ctl.get(); //获取ctl的低29位 if (workerCountOf(c) < corePoolSize) { //workerCountOf方法根据ctl的低29位,得到线程池的当前线程数,如果当前的线程数小于核心池大小,则直接addWorker if (addWorker(command, true)) return; c = ctl.get(); } if (isRunning(c) && workQueue.offer(command)) { //如果当前线程池处于Running状态(即能够接受任务,处理阻塞队列中的任务),且把任务提交给了阻塞队列 int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) //再次检查线程池的状态,如果线程池没有RUNNING(即无法放入队列),并且将任务从队列移除,交给reject处理 reject(command); else if (workerCountOf(recheck) == 0) //否则,如果线程池数为0,则添加worker,启动一个null任务的线程 addWorker(null, false); } else if (!addWorker(command, false)) //否则如果添加addWork失败,则reject。 reject(command); }
addWorker主要负责创建新的线程并执行任务,主代码如下:
boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // Recheck while holding lock. // Back out on ThreadFactory failure or if // shut down before lock acquired. int rs = runStateOf(ctl.get()); if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); workers.add(w); int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted;
1、继承了AQS类,可以方便的实现工作线程的中止操作;
2、实现了Runnable接口,可以将自身作为一个任务在工作线程中执行;
3、当前提交的任务firstTask作为参数传入Worker的构造方法;
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } /** Delegates main run loop to outer runWorker */ public void run() { runWorker(this); }
线程工厂在创建线程thread时,将Woker实例本身this作为参数传入,当执行start方法启动线程thread时,本质是执行了Worker的runWorker方法。
在线程池执行任务的时候,除了executor还有一种方法叫做submit,通过ExecutorService.submit()方法提交的任务,可以获取任务执行完的返回值。
1、Callable接口类似于Runnable,只是Runnable没有返回值。
2、Callable任务除了返回正常结果之外,如果发生异常,该异常也会被返回,即Future可以拿到异步执行任务各种结果;
3、Future.get方法会导致主线程阻塞,直到Callable任务执行完成;
Java 线程池分析