摘要:而且,线程池中的线程并没有睡眠,而是进入了自旋状态。普通的线程被中断会导致线程继续执行,从而方法运行完毕,线程退出。线程死亡超过时间,任务对列没有数据而返回。线程死亡保证了线程池至少留下个线程。
线程在执行任务时,正常的情况是这样的:
Thread t=new Thread(new Runnable() { @Override public void run() { // TODO Auto-generated method stub } }); t.start();
Thread 在初始化的时候传入一个Runnable,以后就没有机会再传入一个Runable了。那么,woker作为一个已经启动的线程。是如何不断获取Runnable的呢?
这个时候可以使用一个包装器,将线程包装起来,在Run方法内部获取任务。
public final class Worker implements Runnable { Thread thread = null; Runnable task; private BlockingQueuequeues; public Worker(Runnable task, BlockingQueue queues) { this.thread = new Thread(this); this.task = task; this.queues = queues; } public void run() { if (task != null) { task.run(); } try { while (true) { task = queues.take(); if (task != null) { task.run(); } } } catch (InterruptedException e) { e.printStackTrace(); } } public void start() { this.thread.start(); } } public class Main { public static void main(String[] args) { BlockingQueue queues=new ArrayBlockingQueue (100); Worker worker=new Worker(new Runnable() { public void run() { System.out.println("hello!!! "); try { Thread.currentThread().sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } } }, queues); worker.start(); for(int i=0;i<100;i++){ queues.offer(new Runnable() { public void run() { System.out.println("hello!!! "); try { Thread.currentThread().sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } } }); } } }
这样我们就简单地实现了一个“线程池”(可以将这个“线程池”改造成官方的模式,不过可以自己尝试一下)。ThreadPool的这种实现模式是并发编程中经典的Cyclic Work Distribution模式。
那么,这种实现的线程池性能如何呢?
由于其任务队列使用的是阻塞队列,在队列内部是自旋的。Reeteenlok是改进的CLH队列。自旋锁会耗费一定CPU的资源,在拥有大量任务执行下的情况下比较有效。而且,线程池中的线程并没有睡眠,而是进入了自旋状态。
如果是不支持超线程的CPU,在同一时刻的确只能处理2个线程,但是并不意味着双核的CPU只能处理两个线程,它可以通过切换上下文来执行多个线程。比如我只有一个大脑,但是我要处理5个人提交的任务,我可以处理完A的事情后,把事情的中间结果保存下,然后再处理B的,然后再读取A的中间结果,处理A的事情。
JDK中的线程池实现分析Woker自身继承了Runnable,并对Thread做了一个包装。Woker代码如下所示:
private final class Worker extends AbstractQueuedSynchronizer implements Runnable { private static final long serialVersionUID = 6138294804551838833L; Runnable firstTask; volatile long completedTasks; Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } public void run() { runWorker(this); } 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) { } } } }
execute(Runnable command)方法内部是这样的:
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); if (workerCountOf(c) < corePoolSize) { 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); else if (workerCountOf(recheck) == 0) addWorker(null, false); } else if (!addWorker(command, false)) reject(command); }
ctl一个合并类型的值。将当前线程数和线程池状态通过数学运算合并到了一个值。具体是如何合并的可以参看一下源码,这里就不叙述了。继续向下走:
if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); }
可以看到,如果当前线程数量小于了核心线程数量corePoolSize,就直接增加线程处理任务。与队列没有关系。但是紧接着又检查了一遍状态,因为在这个过程中,别的线程也可能在添加任务。继续向下走:
if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command); else if (workerCountOf(recheck) == 0) addWorker(null, false); }
可以看到如果线程池是运行态的,就把线程添加到任务队列。workQueue是构造函数传递过来的,可以是有界队列,也可以是无界队列。可以看出来,队列如果是无界的,直接往队列里面添加任务,这个时候,线程池中的线程也不会增加,一直会等于核心线程数。
如果队列是有界的,就尝试直接新增线程处理任务,如果添加任务失败,就调用reject方法来处理添加失败的任务:
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); //如果状态大于SHUTDOWN,不再接受新的任务,直接返回 if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; /**根据core来判断,如果当前线程数量大于corePoolSize或者最大线程数,直接返回。添加任务失败。 **如果队列是有界的或者任务添加到队列失败(参数core是false),那么就会新开一个线程处理业务,但如果线程已经大于了maximumPoolSize,就会出现添加失败,返回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) { t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }
如果创建失败的情况下会调用addWorkerFailed方法,从而将减少实际线程数。
addWorker中for循环的意义在addWorker中有这么一段代码,表示为当前线程数加1:
private boolean compareAndIncrementWorkerCount(int expect) { return ctl.compareAndSet(expect, expect + 1); }
由于多线程可能同时操作。expect值可能会变化。仅仅一次的操作compareAndIncrementWorkerCount可能一次并不会成功,而且,一个线程在执行addWork的过程中间,另外一个线程假设直接shotdown这个线程池。for循环的存在可以保证状态一定是一致的。
任务的执行在Worker中间实际上是调用的runWorker方法来执行的具体业务:
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 ((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); } }
可以看到while循环不断的从队列中取出任务执行。如果task==null 并且getTask()等于null的话,那么就会跳出循环,进入到processWorkerExit,run方法执行完毕以后,这个线程也被销毁了。但是为什么在各自的线程执行,为什么还需要加锁呢?答案是因为要线程池需要判断这个线程是否在执行任务。在interruptIdleWorkers方法中,要中断那写目前空闲的线程,通过当前Worker是否获得了锁就能判断这个worker是否是空闲的:
private void interruptIdleWorkers(boolean onlyOne) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (Worker w : workers) { Thread t = w.thread; if (!t.isInterrupted() && w.tryLock()) { try { t.interrupt(); } catch (SecurityException ignore) { } finally { w.unlock(); //中断不起作用。interrupt()对于自旋锁是不起作用的。只是逻辑上被阻塞, } } if (onlyOne) break; } } finally { mainLock.unlock(); } }
可以看到,如果w.tryLock()可以获取到锁,那么就意味着当前的 Woker并没有处理任务(没有进入到循环里面或者被getTask方法所阻塞,无法获取锁)。
Work之所以继承AbstractQueuedSynchronizer,而不去使用ReentrantLock。是因为ReentrantLock是可重入锁,在调用lock方法获取锁之后,再调用tryLock()还是会返回true。
public static void main(String[] args) { ReentrantLock lock = new ReentrantLock(); lock.lock(); System.out.println(lock.tryLock()); }
输出结果是true,所以使用ReentrantLock则难以判断当前Worker是否在执行任务。
线程超时allowCoreThreadTimeOut、keepAliveTime以及线程死亡在上面的interruptIdleWorkers方法中,线程被中断。普通的线程被中断会导致线程继续执行,从而run方法运行完毕,线程退出。
对于一个没有被阻塞的线程,中断是不起作用的。中断在如下线程被阻塞的方法中起作用:
the wait(),
wait(long),
wait(long, int)
join(),
join(long),
join(long, int),
sleep(long),
or sleep(long, int)
LockSupport.park(Object object);
LockSupport.park();
,如果唤醒这些被阻塞的线程,从而能使得run方法继续执行,当run方法执行完毕,那么线程也就终结死亡。但是对于ReentrantLock和AbstractQueuedSynchronizer这种自旋+CAS实现的“逻辑锁”,是不起作用的。
而且runWork本身也是While循环,靠中断是无法退出循环的。
但是在ThreadPoolExecutor的构造函数中,有一个允许设置线程超时allowCoreThreadTimeOut参数的方法。如果允许超时,多于corePoolSize的线程将会在处在空闲状态之后存活keepAliveTime时长后终止。因此有了一个allowCoreThreadTimeOut方法:
public void allowCoreThreadTimeOut(boolean value) { if (value && keepAliveTime <= 0) throw new IllegalArgumentException("Core threads must have nonzero keep alive times"); if (value != allowCoreThreadTimeOut) { allowCoreThreadTimeOut = value; if (value) interruptIdleWorkers(); } }
正如上面提到的一样,允许allowCoreThreadTimeOut并且调用interruptIdleWorkers方法并不能使线程退出。那么线程池又如何杀掉这个线程呢?
没错,就是getTask方法。只有当getTask返回null的时候才能跳出While循环,run方法运行完毕,那么线程自然而然就死亡了。getTask方法如下所示:
private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? for (;;) { int c = ctl.get(); int rs = runStateOf(c); // Check if queue empty only if necessary. if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); // Are workers subject to culling? boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } try { Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }
可以看到,如果线程池状态大于SHUTDOWN并且队列空,返回null,从而结束循环。(线程死亡)
或者状态大于SHUTDOWN并且线程大于STOP(STOP一定大于SHUTDOWN,所以可以直接说线程大于STOP)返回null,从而结束循环。(线程死亡)
再往下可以看到如果超过了maximumPoolSize,返回null,从而结束循环。(线程死亡)
超过keepAliveTime时间,任务对列没有数据而返回null。从而结束循环。(线程死亡)
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;保证了线程池至少留下corePoolSize个线程。
在execute方法中,如果线程池拒绝添加任务,就会有一个钩子方法来处理被拒绝的任务。
可以自己定义,也可以使用线城池中默认的拒接处理协议。
AbortPolicy :直接抛出RejectedExecutionException异常;
CallerRunsPolicy:谁调用的execute方法,谁就执行这个任务;
DiscardPolicy:直接丢弃,什么也不做;
DiscardOldestPolicy:丢弃对列中间最老的任务,执行新任务。
有什么问题或者建议,可以加入小密圈和我一起讨论,或者在简书留言,欢迎喜欢和打赏。
最后向大家安利一本我写的关于Java并发的书籍:Java并发编程系统与模型,个人觉得写得不错,比较通俗易懂,非常适合初学者,百度阅读可以下载电子书。
文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。
转载请注明本文地址:https://www.ucloud.cn/yun/67354.html
摘要:理解内存模型对多线程编程无疑是有好处的。干货高级动画高级动画进阶,矢量动画。 这是最好的Android相关原创知识体系(100+篇) 知识体系从2016年开始构建,所有的文章都是围绕着这个知识体系来写,目前共收入了100多篇原创文章,其中有一部分未收入的文章在我的新书《Android进阶之光》中。最重要的是,这个知识体系仍旧在成长中。 Android 下拉刷新库,这一个就够了! 新鲜出...
摘要:最后,我们会通过对源代码的剖析深入了解线程池的运行过程和具体设计,真正达到知其然而知其所以然的水平。创建线程池既然线程池是一个类,那么最直接的使用方法一定是一个类的对象,例如。单线程线程池单线程线程 我们一般不会选择直接使用线程类Thread进行多线程编程,而是使用更方便的线程池来进行任务的调度和管理。线程池就像共享单车,我们只要在我们有需要的时候去获取就可以了。甚至可以说线程池更棒,...
摘要:具体调用链路如图函数主要是解析启动参数,并过滤选项传给引擎。查阅文档之后发现,通过指定参数可以设置线程池大小。原来的字节码编译优化还有都是通过多线程完成又继续深入调查,发现环境变量会影响的线程池大小。执行过程如下调用执行。 作者:正龙 (沪江Web前端开发工程师)本文原创,转载请注明作者及出处。 随着Node.js的普及,越来越多的开发者使用Node.js来搭建环境,也有很多公司开始把...
阅读 3702·2021-10-11 10:59
阅读 1253·2019-08-30 15:44
阅读 3465·2019-08-29 16:39
阅读 2869·2019-08-29 16:29
阅读 1783·2019-08-29 15:24
阅读 792·2019-08-29 15:05
阅读 1250·2019-08-29 12:34
阅读 2284·2019-08-29 12:19