线程池

[TOC]

线程池

1. 并发队列：阻塞队列和非阻塞队列

区别如下：

入队：

• 非阻塞队列：当队列中满了的时候，放入数据，数据丢失
• 阻塞队列：当队列满了的时候，进行等待，什么时候队列中有出队的数据，那么第 11 个再放进入

出队：

• 非阻塞队列：如果现在队列中没有元素，取数据，得到的是 null
• 阻塞队列：等待，什么时候放进去，再取出来

线程池是使用阻塞队列。

---

2. 线程池原理：ThreadPoolExecutor 底层原理解析

2. 1线程的生命周期

从出生到死亡的阶段。

2.2 线程池工作原理

ThreadPoolExecutor t = new ThreadPoolExecutor(1, 3, 0, TimeUnit.MICROSECONDS, new LinkedBlockingDeque<>(3));

参数含义：

• 1：核心线程数
• 3：最大线程数
• 0 和 TimeUnit.MICROSECONDS：当任务量大于队列长度需要创建线程的时候，新创建的这个线程假如把队列的任务都执行完了，那么等待新任务的空闲时间就是这个
• new LinkedBlockingDeque<>(3))：阻塞队列：
  • 方法有参数：队列长度为 3
  • 方法无参数：队列长度为 Integer.MAX_VALUE

执行流程图如下所示：

2.3 举个例子

package com.test.jdk8;

import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class TestThreadPoolExecutor {

    public static void main(String[] args) {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 2, 3, TimeUnit.SECONDS, new LinkedBlockingDeque<>(3));
        // 利用线程池里的线程，开始执行任务
        // 执行第一个任务
        pool.execute(new TestThread());
        // 执行下面三个任务，放入阻塞队列，阻塞队列满：
        pool.execute(new TestThread());
        pool.execute(new TestThread());
        pool.execute(new TestThread());

        // 执行第 4个任务，创建新线程，分摊任务
        pool.execute(new TestThread());

        // 执行第 5 个任务，创建新创建，线程数大于最大线程数，报错
        pool.execute(new TestThread());


        // 关闭线程池
        pool.shutdown();
    }

}

class TestThread implements Runnable {
    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName());
    }
}

---

3. 线程池的分类

1. newCachedThreadPool 可缓存线程池

创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。示例代码如下：

package ThreadTest;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Test {

    public static void main(String[] args) {
        ExecutorService es = Executors.newCachedThreadPool();

        for (int i = 0; i < 100; i++) {
            // lambda 表达式, 实现 new Runnable() 的 run() 方法
            es.execute(() -> System.out.println(Thread.currentThread().getName()));
        }

        es.shutdown();

    }

}

查看 newCachedThreadPool 源码，可知：

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

该线程池的核心线程数为 0， 最大线程数为 Integer.MAX_VALUE，将会根据需要创建新线程，新线程的数量不大于 Integer.MAX_VALUE，线程执行任务结束后被释放可重复使用该线程执行另一个任务。

部分执行结果如下：

pool-1-thread-6
pool-1-thread-11
pool-1-thread-6
pool-1-thread-10
pool-1-thread-4
pool-1-thread-11
pool-1-thread-2

2. newFixedThreadPool 定长线程池

创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。示例代码如下：

package ThreadTest;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Test {

    public static void main(String[] args) {
        ExecutorService es = Executors.newFixedThreadPool(3);

        for (int i = 0; i < 100; i++) {
            // lambda 表达式, 实现 new Runnable() 的 run() 方法
            es.execute(() -> System.out.println(Thread.currentThread().getName()));
        }

        es.shutdown();

    }

}

查看 newFixedThreadPool 源码，可知：

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

传入的参数即为核心线程数和最大线程数，核心线程数和最大线程数的数量是相等的。

所以上述示例代码的执行结果的线程数就为 3，部分执行结果如下：

pool-1-thread-1
pool-1-thread-3
pool-1-thread-2
pool-1-thread-3
pool-1-thread-1

3. newScheduledThreadPool 定时线程池

创建一个定时线程池，支持定时及周期性任务执行。延迟执行示例代码如下：

package ThreadTest;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class Test {

    public static void main(String[] args) {
        ScheduledExecutorService ses = Executors.newScheduledThreadPool(2);

        for (int i = 0; i < 100; i++) {
            ses.schedule(() -> System.out.println(Thread.currentThread().getName()), 3, TimeUnit.SECONDS);
        }

        ses.shutdown();

    }
}

查看 newScheduledThreadPool 源码，如下：

public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
    }

可知，newScheduledThreadPool 传入的参数即为核心线程数，最大线程数为 Integer.MAX_VALUE。

上述代码通过设置延迟参数 3， 单位 TimeUnit.SECONDS，延迟三秒后执行任务，创建的线程数为 2。执行部分结果如下：

pool-1-thread-1
pool-1-thread-2
pool-1-thread-1
pool-1-thread-2
pool-1-thread-1

4. newSingleThreadExecutor 单例线程池

创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。示例代码如下：

package ThreadTest;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Test {

    public static void main(String[] args) {
        ExecutorService es = Executors.newSingleThreadExecutor();

        for (int i = 0; i < 100; i++) {
            es.execute(() -> System.out.println(Thread.currentThread().getName()));
        }

        es.shutdown();

    }
}

查看 newSingleThreadExecutor 源码，可知：

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

核心线程数和最大线程数都为 1，执行完任务后线程的存活时间为 0，即创建出来的线程只会执行一次任务，保证单例执行。

运行部分结果如下：

pool-1-thread-1
pool-1-thread-1
pool-1-thread-1