java 中ThreadPoolExecutor原理分析
java 中threadpoolexecutor原理分析
线程池简介
java线程池是开发中常用的工具,当我们有异步、并行的任务要处理时,经常会用到线程池,或者在实现一个服务器时,也需要使用线程池来接收连接处理请求。
线程池使用
jdk中提供的线程池实现位于java.util.concurrent.threadpoolexecutor。在使用时,通常使用executorservice接口,它提供了submit,invokeall,shutdown等通用的方法。
在线程池配置方面,executors类中提供了一些静态方法能够提供一些常用场景的线程池,如newfixedthreadpool,newcachedthreadpool,newsinglethreadexecutor等,这些方法最终都是调用到了threadpoolexecutor的构造函数。
threadpoolexecutor的包含所有参数的构造函数是
/** * @param corepoolsize the number of threads to keep in the pool, even * if they are idle, unless {@code allowcorethreadtimeout} is set * @param maximumpoolsize the maximum number of threads to allow in the * pool * @param keepalivetime when the number of threads is greater than * the core, this is the maximum time that excess idle threads * will wait for new tasks before terminating. * @param unit the time unit for the {@code keepalivetime} argument * @param workqueue the queue to use for holding tasks before they are * executed. this queue will hold only the {@code runnable} * tasks submitted by the {@code execute} method. * @param threadfactory the factory to use when the executor * creates a new thread * @param handler the handler to use when execution is blocked * because the thread bounds and queue capacities are reached 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,则不管当前是否有线程闲置,都会创建一个新的线程来执行任务。
- maximunpoolsize是线程池中允许的最大的线程数
- workqueue用于存放排队的任务
- keepalivetime是大于corepoolsize的线程闲置的超时时间
- handler用于在任务逸出、线程池关闭时的任务处理 ,线程池的线程增长策略为,当前线程数小于corepoolsize时,新增线程,当线程数=corepoolsize且corepoolsize时,只有在workqueue不能存放新的任务时创建新线程,超出的线程在闲置keepalivetime后销毁。
实现(基于jdk1.8)
threadpoolexecutor中保存的状态有
当前线程池状态, 包括running,shutdown,stop,tidying,terminated。
当前有效的运行线程的数量。
将这两个状态放到一个int变量中,前三位作为线程池状态,后29位作为线程数量。
例如0b11100000000000000000000000000001, 表示running, 一个线程。
通过hashset来存储工作者集合,访问该hashset前必须先获取保护状态的mainlock:reentrantlock
submit、execute
execute的执行方式为,首先检查当前worker数量,如果小于corepoolsize,则尝试add一个core worker。线程池在维护线程数量以及状态检查上做了大量检测。
public void execute(runnable command) { int c = ctl.get(); // 如果当期数量小于corepoolsize if (workercountof(c) < corepoolsize) { // 尝试增加worker if (addworker(command, true)) return; c = ctl.get(); } // 如果线程池正在运行并且成功添加到工作队列中 if (isrunning(c) && workqueue.offer(command)) { // 再次检查状态,如果已经关闭则执行拒绝处理 int recheck = ctl.get(); if (! isrunning(recheck) && remove(command)) reject(command); // 如果工作线程都down了 else if (workercountof(recheck) == 0) addworker(null, false); } else if (!addworker(command, false)) reject(command); }
addworker方法实现
private boolean addworker(runnable firsttask, boolean core) { retry: for (;;) { int c = ctl.get(); int rs = runstateof(c); // check if queue empty only if necessary. if (rs >= shutdown && ! (rs == shutdown && firsttask == null && ! workqueue.isempty())) return false; for (;;) { int wc = workercountof(c); if (wc >= capacity || wc >= (core ? corepoolsize : maximumpoolsize)) return false; if (compareandincrementworkercount(c)) break retry; c = ctl.get(); // re-read ctl if (runstateof(c) != rs) continue retry; // else cas failed due to workercount change; retry inner loop } } 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) { // 如果添加成功,则启动该线程,执行worker的run方法,worker的run方法执行外部的runworker(worker) t.start(); workerstarted = true; } } } finally { if (! workerstarted) addworkerfailed(w); } return workerstarted; }
worker类继承了abstractqueuedsynchronizer获得了同步等待这样的功能。
private final class worker extends abstractqueuedsynchronizer implements runnable { /** * this class will never be serialized, but we provide a * serialversionuid to suppress a javac warning. */ private static final long serialversionuid = 6138294804551838833l; /** thread this worker is running in. null if factory fails. */ final thread thread; /** initial task to run. possibly null. */ runnable firsttask; /** per-thread task counter */ volatile long completedtasks; /** * creates with given first task and thread from threadfactory. * @param firsttask the first task (null if none) */ 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); } // lock methods // // the value 0 represents the unlocked state. // the value 1 represents the locked state. protected boolean isheldexclusively() { return getstate() != 0; } protected boolean tryacquire(int unused) { if (compareandsetstate(0, 1)) { setexclusiveownerthread(thread.currentthread()); return true; } return false; } protected boolean tryrelease(int unused) { setexclusiveownerthread(null); setstate(0); return true; } public void lock() { acquire(1); } public boolean trylock() { return tryacquire(1); } public void unlock() { release(1); } public boolean islocked() { return isheldexclusively(); } void interruptifstarted() { thread t; if (getstate() >= 0 && (t = thread) != null && !t.isinterrupted()) { try { t.interrupt(); } catch (securityexception ignore) { } } }
runworker(worker)是worker的轮询执行逻辑,不断地从工作队列中获取任务并执行它们。worker每次执行任务前需要进行lock,防止在执行任务时被interrupt。
final void runworker(worker w) { thread wt = thread.currentthread(); runnable task = w.firsttask; w.firsttask = null; w.unlock(); // allow interrupts boolean completedabruptly = true; try { while (task != null || (task = gettask()) != null) { w.lock(); // if pool is stopping, ensure thread is interrupted; // if not, ensure thread is not interrupted. this // requires a recheck in second case to deal with // shutdownnow race while clearing interrupt if ((runstateatleast(ctl.get(), stop) || (thread.interrupted() && runstateatleast(ctl.get(), stop))) && !wt.isinterrupted()) wt.interrupt(); try { beforeexecute(wt, task); throwable thrown = null; try { task.run(); } catch (runtimeexception x) { thrown = x; throw x; } catch (error x) { thrown = x; throw x; } catch (throwable x) { thrown = x; throw new error(x); } finally { afterexecute(task, thrown); } } finally { task = null; w.completedtasks++; w.unlock(); } } completedabruptly = false; } finally { processworkerexit(w, completedabruptly); } }
threadpoolexecutor的submit方法中将callable包装成futuretask后交给execute方法。
futuretask
futuretask继承于runnable和future,futuretask定义的几个状态为
new, 尚未执行
completing, 正在执行
normal, 正常执行完成得到结果
exceptional, 执行抛出异常
cancelled, 执行被取消
interrupting,执行正在被中断
interrupted, 已经中断。
其中关键的get方法
public v get() throws interruptedexception, executionexception { int s = state; if (s <= completing) s = awaitdone(false, 0l); return report(s); }
先获取当前状态,如果还未执行完成并且正常,则进入等待结果流程。在awaitdone不断循环获取当前状态,如果没有结果,则将自己通过cas的方式添加到等待链表的头部,如果设置了超时,则locksupport.parknanos到指定的时间。
static final class waitnode { volatile thread thread; volatile waitnode next; waitnode() { thread = thread.currentthread(); } } private int awaitdone(boolean timed, long nanos) throws interruptedexception { final long deadline = timed ? system.nanotime() + nanos : 0l; waitnode q = null; boolean queued = false; for (;;) { if (thread.interrupted()) { removewaiter(q); throw new interruptedexception(); } int s = state; if (s > completing) { if (q != null) q.thread = null; return s; } else if (s == completing) // cannot time out yet thread.yield(); else if (q == null) q = new waitnode(); else if (!queued) queued = unsafe.compareandswapobject(this, waitersoffset, q.next = waiters, q); else if (timed) { nanos = deadline - system.nanotime(); if (nanos <= 0l) { removewaiter(q); return state; } locksupport.parknanos(this, nanos); } else locksupport.park(this); } }
futuretask的run方法是执行任务并设置结果的位置,首先判断当前状态是否为new并且将当前线程设置为执行线程,然后调用callable的call获取结果后设置结果修改futuretask状态。
public void run() { if (state != new || !unsafe.compareandswapobject(this, runneroffset, null, thread.currentthread())) return; try { callable<v> c = callable; if (c != null && state == new) { v result; boolean ran; try { result = c.call(); ran = true; } catch (throwable ex) { result = null; ran = false; setexception(ex); } if (ran) set(result); } } finally { // runner must be non-null until state is settled to // prevent concurrent calls to run() runner = null; // state must be re-read after nulling runner to prevent // leaked interrupts int s = state; if (s >= interrupting) handlepossiblecancellationinterrupt(s); } }
感谢阅读,希望能帮助到大家,谢谢大家对本站的支持!
上一篇: java中成员变量与局部变量区别分析