Принцип реализации пула потоков

задняя часть

Преимущества пулов потоков

1. Потоки — дефицитный ресурс.Использование пулов потоков может уменьшить количество создаваемых и уничтожаемых потоков, а каждый рабочий поток можно использовать повторно. Чтобы установить количество потоков, см. введение в эту статью на Geek Time:time.geekgang.org/column/areti…, интенсивный процессор: количество ядер процессора + 1; интенсивный ввод-вывод: одноядерный: 1 + (время ввода-вывода / время процессора), многоядерный: ядра процессора * [ 1 + (время ввода-вывода / время процессора) ]

2. Количество рабочих потоков в пуле потоков можно настроить в соответствии с производительностью системы, чтобы предотвратить сбой сервера из-за чрезмерного потребления памяти.

Создание пула потоков

public ThreadPoolExecutor(int corePoolSize,
                                int maximumPoolSize,
                                long keepAliveTime,
                                TimeUnit unit,
                                BlockingQueue<Runnable> workQueue,
                                RejectedExecutionHandler handler) 

corePoolSize：线程池核心线程数量

maximumPoolSize:线程池最大线程数量

keepAliverTime：当活跃线程数大于核心线程数时，空闲的多余线程最大存活时间

unit：存活时间的单位

workQueue：存放任务的队列

handler：超出线程范围和队列容量的任务的处理程序

注：关于workQueue参数的取值,JDK提供了4种阻塞队列类型供选择：
            ArrayBlockingQueue：基于数组结构的有界阻塞队列，按FIFO排序任务；
            
            inkedBlockingQuene：基于链表结构的阻塞队列，按FIFO排序任务，吞吐量通常要高于ArrayBlockingQuene 

            SynchronousQuene：一个不存储元素的阻塞队列，每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于ArrayBlockingQuene；

            PriorityBlockingQuene：具有优先级的无界阻塞队列；

     threadFactory：线程工厂，主要用来创建线程；

     handler：表示当拒绝处理任务时的策略，有以下四种取值

 注： 当线程池的饱和策略，当阻塞队列满了，且没有空闲的工作线程，如果继续提交任务，必须采取一种策略处理该任务，线程池提供了4种策略：

        ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。

        ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。

        ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）

        ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

        当然也可以根据应用场景实现RejectedExecutionHandler接口，自定义饱和策略，如记录日志或持久化存储不能处理的任务。

Принцип реализации пула потоков

提交一个任务到线程池中，线程池的处理流程如下：

1、判断线程池里的核心线程是否都在执行任务，如果不是（核心线程空闲或者还有核心线程没有被创建）则创建一个新的工作线程来执行任务。如果核心线程都在执行任务，则进入下个流程。

2、线程池判断工作队列是否已满，如果工作队列没有满，则将新提交的任务存储在这个工作队列里。如果工作队列满了，则进入下个流程。

3、判断线程池里的线程是否都处于工作状态，如果没有，则创建一个新的工作线程来执行任务。如果已经满了，则交给饱和策略来处理这个任务。

Интерпретация исходного кода пула потоков

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
            //如果线程数大于等于基本线程数或者线程创建失败，将任务加入队列
        if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
           //线程池处于运行状态并且加入队列成功
            if (runState == RUNNING && workQueue.offer(command)) {
                if (runState != RUNNING || poolSize == 0)
                    ensureQueuedTaskHandled(command);
            }
           //线程池不处于运行状态或者加入队列失败，则创建线程（创建的是非核心线程）
            else if (!addIfUnderMaximumPoolSize(command))
          //创建线程失败，则采取阻塞处理的方式
                reject(command); // is shutdown or saturated
        }
    }

Инициализировать четыре типа пулов потоков

1. новый пул фиксированных потоков ()

说明：初始化一个指定线程数的线程池，其中corePoolSize == maxiPoolSize，使用LinkedBlockingQuene作为阻塞队列
特点：即使当线程池没有可执行任务时，也不会释放线程。

2. новыйCachedThreadPool()

说明：初始化一个可以缓存线程的线程池，默认缓存60s，线程池的线程数可达到Integer.MAX_VALUE，即2147483647，内部使用SynchronousQueue作为阻塞队列；
特点：在没有任务执行时，当线程的空闲时间超过keepAliveTime，会自动释放线程资源；当提交新任务时，如果没有空闲线程，则创建新线程执行任务，会导致一定的系统开销；
因此，使用时要注意控制并发的任务数，防止因创建大量的线程导致而降低性能。

3. newSingleThreadExecutor()

说明：初始化只有一个线程的线程池，内部使用LinkedBlockingQueue作为阻塞队列。
特点：如果该线程异常结束，会重新创建一个新的线程继续执行任务，唯一的线程可以保证所提交任务的顺序执行

4. новый запланированный пул потоков ()

特定：初始化的线程池可以在指定的时间内周期性的执行所提交的任务，在实际的业务场景中可以使用该线程池定期的同步数据。

总结：除了newScheduledThreadPool的内部实现特殊一点之外，其它线程池内部都是基于ThreadPoolExecutor类（Executor的子类）实现的。

Отправить задачу в пул потоков

Есть два способа:

  Executor.execute(Runnable command);

  ExecutorService.submit(Callable<T> task);

Внутренняя реализация execute()

1.首次通过workCountof()获知当前线程池中的线程数，

  如果小于corePoolSize, 就通过addWorker()创建线程并执行该任务；

&emsp;否则，将该任务放入阻塞队列；

2. 如果能成功将任务放入阻塞队列中,  

如果当前线程池是非RUNNING状态，则将该任务从阻塞队列中移除，然后执行reject()处理该任务；

如果当前线程池处于RUNNING状态，则需要再次检查线程池（因为可能在上次检查后，有线程资源被释放），是否有空闲的线程；如果有则执行该任务；

3、如果不能将任务放入阻塞队列中,说明阻塞队列已满；那么将通过addWoker()尝试创建一个新的线程去执行这个任务；如果addWoker()执行失败，说明线程池中线程数达到maxPoolSize,则执行reject()处理任务；

Внутренняя реализация sumbit()

会将提交的Callable任务会被封装成了一个FutureTask对象

FutureTask类实现了Runnable接口，这样就可以通过Executor.execute()提交FutureTask到线程池中等待被执行，最终执行的是FutureTask的run方法； 

比较：

 两个方法都可以向线程池提交任务，execute()方法的返回类型是void，它定义在Executor接口中, 而submit()方法可以返回持有计算结果的Future对象，它定义在ExecutorService接口中，它扩展了Executor接口，其它线程池类像ThreadPoolExecutor和ScheduledThreadPoolExecutor都有这些方法。

жизненный цикл нити

Когда поток создается и запускается, он не переходит в состояние выполнения сразу после запуска и не всегда находится в состоянии выполнения.В жизненном цикле потока он проходит через пять состояний: новый, готовый, работающий , заблокирован и мертв.

новое состояние

Когда программа использует ключевое слово new для создания потока, поток находится в новом состоянии.В это время только JVMJ выделяет для него память и инициализирует значения его переменных-членов.

состояние готовности

Когда объект потока вызывает метод start(), поток находится в состоянии готовности. Виртуальная машина Java создает для него стек вызовов методов и программный счетчик, ожидая запланированного запуска.

Рабочий статус

Если поток в состоянии готовности получает процессор и начинает выполнять выполнение метода run(), поток находится в состоянии выполнения.

состояние блокировки

Блокирующее состояние означает, что поток по какой-либо причине отказывается от права на использование процессора, то есть он отказывается от кванта времени процессора, приостанавливается и прекращает работу. процессорное время снова и перейти в рабочее состояние. переход между состояниями потока

Категории