简单的说就是,我预计最少需要3个(corePoolSize)正事员工(线程),如果事情做不过来,我可以扩招到10个(maximumPoolSize),再多我财务承受不起,只能让在职的人加班,如果还是做不过来,作为老板的我就得定个策略:是帮他们排期呢?还是直接拒绝呢。。。员工呢,签定合同的周期为keepAliveTime ,单位可以是TimeUnit;更有甚着,员工必须是 ThreadFactory 这个地方出来的。 以上是线程池的核心思想,接下来就要深入代码了: //首先是员工: //这是员工列表 private final HashSet<Worker> workers = new HashSet<Worker>(); //这是员工的基本定义 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) { } } } } //这是添加员工的方法 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) { //这点是在容易漏掉 启动一个线程 t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; } //接下来是任务: //其实任务队列是个阻塞队列 private final BlockingQueue<Runnable> workQueue; //然后是获取任务的方法: 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; } } } //执行任务: 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 { //容易搞混的点,thread中start()是开启一个线程,而这里仅仅只是调用run方法。 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); } } //最后可能少一个大家经常看到的入口 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(); 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); }
相关推荐
基于java的开发源码-github-Java-api.zip 基于java的开发源码-github-Java-api.zip 基于java的开发源码-github-Java-api.zip 基于java的开发源码-github-Java-api.zip 基于java的开发源码-github-Java-api.zip 基于...
线程池管理源码 线程池管理源码 线程池管理源码
Java 3DMenu 界面源码,有人说用到游戏中不错,其实平时我信编写Java应用程序时候也能用到吧,不一定非要局限于游戏吧,RES、SRC资源都有,都在压缩包内。 Java zip压缩包查看程序源码 1个目标文件 摘要:Java源码...
源码-java网络爬虫源码 源码-java网络爬虫源码 源码-java网络爬虫源码 源码-java网络爬虫源码 源码-java网络爬虫源码 源码-java网络爬虫源码 源码-java网络爬虫源码 源码-java网络爬虫源码 源码-java网络爬虫源码 ...
java 线程池 完整 源码 java 线程池 完整 源码
java线程池的源码分析以及各种池之间的对比;
JAVA实现的线程池,可以直接在正规大型项目中套用。如果有不懂的,可以问我,QQ:452242541,加时注明csdn
NULL 博文链接:https://shaojiashuai123456.iteye.com/blog/1415886
1.媲美java线程池框架,整套源码资源,使用Intellij Idea开发工具,JDK1.8以上 2.带有测试代码 3.可以根据项目实际情况任意调整代码 4.任务队列、拒绝策略 5.BasicThreadPool.java、LinkedRunnableQueue.java、...
java 游戏 源码 模拟钢琴 你的选择没错
Reference: 《创建Java线程池》[1],《Java线程:新特征-线程池》[2], 《Java线程池学习》[3],《线程池ThreadPoolExecutor使用简介》[4],《Java5中的线程池实例讲解》[5],《ThreadPoolExecutor使用和思考》[6] ...
CoAP实现的Java源码- californium-1.0.0-M3(2015年1月16日最新版本) 另上传C语言实现的libcoap源码。
java线程池源码 cThreadPool 项目描述:对java.util.concurrent包下线程池相关源码进行重新实现,深入研究和学习线程池超时机制、饱和策略、生命周期等知识 ThreadPoolExecutor类下部分方法和内部类介绍: 1、Worker...
java thread类源码 thread-pool-source-code JAVA线程池源码解读
NULL 博文链接:https://lhgyy00.iteye.com/blog/450628
mysql-connector-java-5.1.37jar包和源码下载 最新mysqljar包
面试官系统精讲Java源码及大厂真题最新上传,涉及到的 Java 知识非常齐全,主要有基础类、集合类、并发集合类、队列、线程、锁、线程池、Lambda 流等内容,内容几乎涵盖了工作和面试的核心,如果你对这些源码都...
Java课程设计-基于Java GUI的图书管理系统源码 Java课程设计-基于Java GUI的图书管理系统源码 Java课程设计-基于Java GUI的图书管理系统源码 Java课程设计-基于Java GUI的图书管理系统源码 Java课程设计-基于Java ...
超市系统源码--小型JAVA开发-超市系统后台源码 超市系统源码--小型JAVA开发-超市系统后台源码 超市系统源码--小型JAVA开发-超市系统后台源码 超市管理系统源码--JAVA开发-超市系统后台源码
课程作业-基于C语言实现线程池源码.zip课程作业-基于C语言实现线程池源码.zip课程作业-基于C语言实现线程池源码.zip课程作业-基于C语言实现线程池源码.zip课程作业-基于C语言实现线程池源码.zip课程作业-基于C语言...