线程池解析

2021-12-15 15:29:29 浏览数 (1)

1.1 构造方法：七个参数的分别代表corePoolSize 核心线程数，maximumPoolSize，最大线程数，keepAliveTime

存活时间（也就是线程池中线程可以空闲的时间），unit存活时间的单位，workQueue阻塞队列，threadFactory线程工厂，handler拒绝策略.

代码语言：txt复制
  public ThreadPoolExecutor(int corePoolSize,
代码语言：txt复制
                              int maximumPoolSize,
代码语言：txt复制
                              long keepAliveTime,
代码语言：txt复制
                              TimeUnit unit,
代码语言：txt复制
                              BlockingQueue<Runnable> workQueue,
代码语言：txt复制
                              ThreadFactory threadFactory,
代码语言：txt复制
                              RejectedExecutionHandler handler)

1.2 线程池提交任务的方式有两种

public void execute(Runnable command)，该方法是由Excutor顶层接口提供的

public Future<?> submit(Runnable task) ，该方法是由ExecutorService扩展接口提供。

由返回值类型可以看出一个可以异步的获取返回值结果。

  1. submit方法();该方法将task包装成一个RunnableFuture对象然后调用execute方法执行。
代码语言：txt复制
    public Future<?> submit(Runnable task) {
代码语言：txt复制
        if (task == null) throw new NullPointerException();
代码语言：txt复制
        RunnableFuture<Void> ftask = newTaskFor(task, null);
代码语言：txt复制
        execute(ftask);
代码语言：txt复制
        return ftask;
代码语言：txt复制
    }
  1. execute()方法undefined判断任务是否为空 如果是口则抛出空指针异常。undefined然后获取当前线程池中线程的数量（线程池中的数量是通过成员变量原子整形变量ctl记录，分别用高3位记录线程池的状态和和低29位记录线程池中线程的个数）

线程池的状态undefined 2.1. RUNNING，处于正常运行状态接受任务处理任务undefined 2.2. SHUTDOWN，调用shutdown()方法后进入此状态，表示不在接受新任务，处理阻塞队列中的任务undefined 2.3. STOP，调用shutdownNow()方法后进入此状态不接受新任务，不处理已添加的任，并且会中断正在处理的任务。undefined 2.4. TIDYING， 当所有的任务已终止，ctl记录的”任务数量”为0，线程池会变为TIDYING状态，当线程池变为TIDYING状态时，会执行钩子函数terminated()。terminated()在ThreadPoolExecutor类中是空的，若用户想在线程池变为TIDYING时，进行相应的处理；可以通过重载terminated()函数来实现。undefined 2.5. TERMINATED，线程池彻底终止，就变成TERMINATED状态。

进入具体的执行流程：

判断当前线程池中的线程数量是否小于核心线程数，如果小于则将创建工作线程Woker执行任务，否则继续进行判断当前线程池是否处于运行状态，如果运行则将当前任务加入阻塞队列（这里注意，如果阻塞队列是无界的则最大线程数也没有意义了因为会一直往队列里添加任务）进行等待，否则进行添加非核心线程addWorker(command,

false)fasle表示是非核心线程，具体逻辑在addWorker方法实现

代码语言：txt复制
public void execute(Runnable command) {
代码语言：txt复制
        if (command == null)
代码语言：txt复制
            throw new NullPointerException();
代码语言：txt复制
        int c = ctl.get();
代码语言：txt复制
        if (workerCountOf(c) < corePoolSize) {
代码语言：txt复制
            if (addWorker(command, true))
代码语言：txt复制
                return;
代码语言：txt复制
            c = ctl.get();
代码语言：txt复制
        }
代码语言：txt复制
        if (isRunning(c) && workQueue.offer(command)) {
代码语言：txt复制
            int recheck = ctl.get();
代码语言：txt复制
            if (! isRunning(recheck) && remove(command))
代码语言：txt复制
                reject(command);
代码语言：txt复制
            else if (workerCountOf(recheck) == 0)
代码语言：txt复制
                addWorker(null, false);
代码语言：txt复制
        }
代码语言：txt复制
        else if (!addWorker(command, false))
代码语言：txt复制
            reject(command);
代码语言：txt复制
    }

addWorker()方法

core表示所添加的任务是否是核心线程

首先进入第一个for(;;)进入执行进行状态检查，然后进入第二个for(;;)使用CAS的方式将线程池中线程数量加1 ， wc >= (core ?

corePoolSize : maximumPoolSize))这句判断为区分核心线程数与非核心线程数的依据。

接下来进入第二个大逻辑块，先将当前线程包装为Worker对象(Woker为继承CAS类与实现runnable接口)，获取到当前Worker的真正的处理线程，如果不为Null则获取全局锁进行添加工作线程，进入同步代码块判断线程池状态

wokers为当前线程池中所有工作线程池集合是一个HashSet，将工作线程添加到HashSet之后进行解锁，然后启动工作线程进入runWorker方法（）

代码语言：txt复制
private boolean addWorker(Runnable firstTask, boolean core) {
代码语言：txt复制
        retry:
代码语言：txt复制
        for (;;) {
代码语言：txt复制
            int c = ctl.get();
代码语言：txt复制
            int rs = runStateOf(c);
代码语言：txt复制
            if (rs >= SHUTDOWN &&
代码语言：txt复制
                ! (rs == SHUTDOWN &&
代码语言：txt复制
                   firstTask == null &&
代码语言：txt复制
                   ! workQueue.isEmpty()))
代码语言：txt复制
                return false;
代码语言：txt复制
            for (;;) {
代码语言：txt复制
                int wc = workerCountOf(c);
代码语言：txt复制
                if (wc >= CAPACITY ||
代码语言：txt复制
                    wc >= (core ? corePoolSize : maximumPoolSize))
代码语言：txt复制
                    return false;
代码语言：txt复制
                if (compareAndIncrementWorkerCount(c))
代码语言：txt复制
                    break retry;
代码语言：txt复制
                c = ctl.get();  // Re-read ctl
代码语言：txt复制
                if (runStateOf(c) != rs)
代码语言：txt复制
                    continue retry;
代码语言：txt复制
            }
代码语言：txt复制
        }
代码语言：txt复制
        boolean workerStarted = false;
代码语言：txt复制
        boolean workerAdded = false;
代码语言：txt复制
        Worker w = null;
代码语言：txt复制
        try {
代码语言：txt复制
            w = new Worker(firstTask);
代码语言：txt复制
            final Thread t = w.thread;
代码语言：txt复制
            if (t != null) {
代码语言：txt复制
                final ReentrantLock mainLock = this.mainLock;
代码语言：txt复制
                mainLock.lock();
代码语言：txt复制
                try {
代码语言：txt复制
                    int rs = runStateOf(ctl.get());
代码语言：txt复制
                    if (rs < SHUTDOWN ||
代码语言：txt复制
                        (rs == SHUTDOWN && firstTask == null)) {
代码语言：txt复制
                        if (t.isAlive()) // precheck that t is startable
代码语言：txt复制
                            throw new IllegalThreadStateException();
代码语言：txt复制
                        workers.add(w);
代码语言：txt复制
                        int s = workers.size();
代码语言：txt复制
                        if (s > largestPoolSize)
代码语言：txt复制
                            largestPoolSize = s;
代码语言：txt复制
                        workerAdded = true;
代码语言：txt复制
                    }
代码语言：txt复制
                } finally {
代码语言：txt复制
                    mainLock.unlock();
代码语言：txt复制
                }
代码语言：txt复制
                if (workerAdded) {
代码语言：txt复制
                    t.start();
代码语言：txt复制
                    workerStarted = true;
代码语言：txt复制
                }
代码语言：txt复制
            }
代码语言：txt复制
        } finally {
代码语言：txt复制
            if (! workerStarted)
代码语言：txt复制
                addWorkerFailed(w);
代码语言：txt复制
        }
代码语言：txt复制
        return workerStarted;
代码语言：txt复制
    }

Worker类的runWorker方法(),首先进行解锁，这时候如果调用shutdown方法的话是允许被中断的，然后进入while循环判断当前任务是否为空如果为空则从任务队列中取元素，获取Woker上的锁，获取到之后进入任务的具体执行如果调用shutdown方法的话是不允许被中断的，判断线程池状态如果是stop则中断当前线程，否则执行任务的具体流程，执行完毕之后将task置空，将当前工作线程处理的任务数加一，解锁，继续调用gettask方法执行取任务执行任务的流程，退出while循环的条件为没有取到任务则退出completedAbruptly代表是抛出异常退出(true)还是正常退出（fasle）

代码语言：txt复制
final void runWorker(Worker w) {
代码语言：txt复制
        Thread wt = Thread.currentThread();
代码语言：txt复制
        Runnable task = w.firstTask;
代码语言：txt复制
        w.firstTask = null;
代码语言：txt复制
        w.unlock(); // allow interrupts
代码语言：txt复制
        boolean completedAbruptly = true;
代码语言：txt复制
        try {
代码语言：txt复制
            while (task != null || (task = getTask()) != null) {
代码语言：txt复制
                w.lock();
代码语言：txt复制
                if ((runStateAtLeast(ctl.get(), STOP) ||
代码语言：txt复制
                     (Thread.interrupted() &&
代码语言：txt复制
                      runStateAtLeast(ctl.get(), STOP))) &&
代码语言：txt复制
                    !wt.isInterrupted())
代码语言：txt复制
                    wt.interrupt();
代码语言：txt复制
                try {
代码语言：txt复制
                    beforeExecute(wt, task);
代码语言：txt复制
                    Throwable thrown = null;
代码语言：txt复制
                    try {
代码语言：txt复制
                        task.run();
代码语言：txt复制
                    } catch (RuntimeException x) {
代码语言：txt复制
                        thrown = x; throw x;
代码语言：txt复制
                    } catch (Error x) {
代码语言：txt复制
                        thrown = x; throw x;
代码语言：txt复制
                    } catch (Throwable x) {
代码语言：txt复制
                        thrown = x; throw new Error(x);
代码语言：txt复制
                    } finally {
代码语言：txt复制
                        afterExecute(task, thrown);
代码语言：txt复制
                    }
代码语言：txt复制
                } finally {
代码语言：txt复制
                    task = null;
代码语言：txt复制
                    w.completedTasks  ;
代码语言：txt复制
                    w.unlock();
代码语言：txt复制
                }
代码语言：txt复制
            }
代码语言：txt复制
            completedAbruptly = false;
代码语言：txt复制
        } finally {
代码语言：txt复制
            processWorkerExit(w, completedAbruptly);
代码语言：txt复制
        }
代码语言：txt复制
    }

getTask()方法

timeOut代表是否获取超时

进入循环进行从任务队列中取任务，首先判断当前线程池的状态根据其他代码逻辑可知，如果处于shutdown状态并且任务队列为空的情况下则将当前工作线程数CAS减1然后return

null将没有取代任务，如果处于stop状态则将当前线程池工作线程数减以不管任务队列是否还有任务。

如果线程池处于running状态，则获取当前线程池中线程的数量

然后用变量time标记为当前线程池线程是否设置超时时间或者当前线程数是否大于核心线程数。进行线程池中线程数量逻辑的判断，之后从任务队列里面取任务分为阻塞等待和定时将取到任务返回

代码语言：txt复制
 private Runnable getTask() {
代码语言：txt复制
        boolean timedOut = false; // Did the last poll() time out?
代码语言：txt复制
        for (;;) {
代码语言：txt复制
            int c = ctl.get();
代码语言：txt复制
            int rs = runStateOf(c);
代码语言：txt复制
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
代码语言：txt复制
                decrementWorkerCount();
代码语言：txt复制
                return null;
代码语言：txt复制
            }
代码语言：txt复制
            int wc = workerCountOf(c)
代码语言：txt复制
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
代码语言：txt复制
            if ((wc > maximumPoolSize || (timed && timedOut))
代码语言：txt复制
                && (wc > 1 || workQueue.isEmpty())) {
代码语言：txt复制
                if (compareAndDecrementWorkerCount(c))
代码语言：txt复制
                    return null;
代码语言：txt复制
                continue;
代码语言：txt复制
            }
代码语言：txt复制
            try {
代码语言：txt复制
                Runnable r = timed ?
代码语言：txt复制
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
代码语言：txt复制
                    workQueue.take();
代码语言：txt复制
                if (r != null)
代码语言：txt复制
                    return r;
代码语言：txt复制
                timedOut = true;
代码语言：txt复制
            } catch (InterruptedException retry) {
代码语言：txt复制
                timedOut = false;
代码语言：txt复制
            }
代码语言：txt复制
        }
代码语言：txt复制
    }

processWorkerExit方法

为将去除工作线程后的处理工作，判断当前工作线程是否正常退出，如果因为异常退出则将线程数减1，然后获取全局锁，将该工作线程处理的任务数量加到总任务数量之中，从HashSet移除当前工作线程，解锁，尝试停止线程池。

代码语言：txt复制
private void processWorkerExit(Worker w, boolean completedAbruptly) {
代码语言：txt复制
        if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
代码语言：txt复制
            decrementWorkerCount();
代码语言：txt复制
        final ReentrantLock mainLock = this.mainLock;
代码语言：txt复制
        mainLock.lock();
代码语言：txt复制
        try {
代码语言：txt复制
            completedTaskCount  = w.completedTasks;
代码语言：txt复制
            workers.remove(w);
代码语言：txt复制
        } finally {
代码语言：txt复制
            mainLock.unlock();
代码语言：txt复制
        }
代码语言：txt复制
        tryTerminate();
代码语言：txt复制
        int c = ctl.get();
代码语言：txt复制
        if (runStateLessThan(c, STOP)) {
代码语言：txt复制
            if (!completedAbruptly) {
代码语言：txt复制
                int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
代码语言：txt复制
                if (min == 0 && ! workQueue.isEmpty())
代码语言：txt复制
                    min = 1;
代码语言：txt复制
                if (workerCountOf(c) >= min)
代码语言：txt复制
                    return; // replacement not needed
代码语言：txt复制
            }
代码语言：txt复制
            addWorker(null, false);
代码语言：txt复制
        }
代码语言：txt复制
    }

shutdown()方法在获取执行的时候会先要获取到Woker的锁然后再去工作线程

shutdownnow（）则是直接进行中断不需要获取锁

java提供了Executors线程池的工具类

图片.png

  1. 固定线程数量线程池undefined该线程池核心线程数与最大线程数数量相等，且工作线程的存活时间为0，阻塞队列使用的是无界阻塞队列基于链表，适用于可以预测线程数量的业务，或者服务器负载较重，对当前线程数量进行控制。
代码语言：txt复制
public static ExecutorService newFixedThreadPool(int nThreads) {
代码语言：txt复制
        return new ThreadPoolExecutor(nThreads, nThreads,
代码语言：txt复制
                                      0L, TimeUnit.MILLISECONDS,
代码语言：txt复制
                                      new LinkedBlockingQueue<Runnable>());
代码语言：txt复制
    }
  1. 单例线程池undefined核心线程数与最大线程数都是1 ，其实最大线程数没有意义因为使用的是无界阻塞队列，所有任务都是按顺序执行的
代码语言：txt复制
    public static ExecutorService newSingleThreadExecutor() {
代码语言：txt复制
        return new FinalizableDelegatedExecutorService
代码语言：txt复制
            (new ThreadPoolExecutor(1, 1,
代码语言：txt复制
                                    0L, TimeUnit.MILLISECONDS,
代码语言：txt复制
                                    new LinkedBlockingQueue<Runnable>()));
代码语言：txt复制
    }
  1. 创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲空间，如无可回重用时则创建新线程,队列采用的同步队列，只有当一个任务被取走的时候，才能加入下一个任务，类似于生产者消费者模式
代码语言：txt复制
    public static ExecutorService newCachedThreadPool() {
代码语言：txt复制
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
代码语言：txt复制
                                      60L, TimeUnit.SECONDS,
代码语言：txt复制
                                      new SynchronousQueue<Runnable>());
代码语言：txt复制
    }
  1. 定时任务线程池使用延迟队列，可以实现任务的定时执行。
代码语言：txt复制
    public ScheduledThreadPoolExecutor(int corePoolSize) {
代码语言：txt复制
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
代码语言：txt复制
              new DelayedWorkQueue());
代码语言：txt复制
    }

5.:创建一个拥有多个任务队列的线程池，可以减少连接数创建当前可用cpu数量的线程来并行执行，适用于大耗时的操作，可以并行来执行，底层使用forkjoin框架，实现任务窃取

代码语言：txt复制
    public static ExecutorService newWorkStealingPool(int parallelism) {
代码语言：txt复制
        return new ForkJoinPool
代码语言：txt复制
            (parallelism,
代码语言：txt复制
             ForkJoinPool.defaultForkJoinWorkerThreadFactory,
代码语言：txt复制
             null, true);
代码语言：txt复制
    }
hashset shutdown task undefined 线程池

0 人点赞