jdk 源码分析(13)java ExecutorService 和Executors

xiaoxiao2021-02-27  242

1)Executors: Executors 是线程池的工厂类,主要生产 ExecutorService 和callable 常用的线程池: newFixedThreadPool newSingleThreadExecutor newCachedThreadPool newScheduledThreadPool 不管是定义了那种名字 ,底层都是ThreadPoolExecutor或者其子类。 1) newFixedThreadPool:大小固定, 2) newSingleThreadExecutor里面就一个线程 3) newCachedThreadPool线程会在一段时间内如果没有使用将消失 4) newScheduledThreadPool定时执行一下里面的线程 2)ExecutorService  线程实现:ThreadPoolExecutor 在现场池里都有一个缓冲队列,存储当前线程个数。 private final BlockingQueue<Runnable> workQueue; 添加线程: 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); } 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)) //启动的线程增加1 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; } 这里的Thread  t = w.thread 和t.start ,启动的其实是包装线程。 Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } public void run() { runWorker(this); } 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 {                    //注意,注意,注意,这里调用的是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); } } 总结: 线程池首先判断有多少可以运行, 1)如果可以运行,直接加入到workers 集合中,并且里面启动包装类worker,worker,在执行完当前线程后会去线程池中寻找下一个线程,并且执行 2)如果当前执行的线程大于设定的数目,就线程放入缓存队列queue 里面。这里的线程,会在上(1)中执行后调用。
转载请注明原文地址: https://www.6miu.com/read-10051.html

最新回复(0)