多线程与高并发——并发编程（5）

五、线程池

1 什么是线程池

为什么要使用线程池？

• 在开发中，为了提升效率，我们需要将一些业务采用多线程的方式去执行。比如，有一个比较大的任务，可以将任务分成几块，分别交给几个线程去执行，最终做一个汇总即可。再比如，做业务操作时，需要发送短信或邮件，这些操作也可以基于异步的方式完成，这种异步的方式，其实就是在构建一个线程去执行。
• 但是，如果每次异步操作或者多线程操作都需要新创建一个线程，使用完毕后，线程再被销毁。这样的话，会对系统造成一些额外的开销。在处理过程中，到底有多少线程处理了多少任务，以及每个线程的开销也无法统计和管理，所以我们需要一个线程池机制来管理这些内容。
• 线程池的概念和连接池类似，就是在一个Java的集合中存储大量的线程对象，每次需要执行异步操作或者多线程操作时，不需要重新创建线程，直接从集合中拿到线程对象直接执行方法就可以了。
• JDK中提供了线程池的类。在线程池构建的初期，可以将任务提交到线程池中，会根据一定的机制来异步执行这个任务。
  • 可能任务直接被执行；
  • 任务可以暂存起来，等到有空闲线程再来处理；
  • 任务也可能被拒绝，无法被执行。
• JDK提供的线程池记录了每个线程处理了多少个任务，以及整个线程池处理了多少个任务。同时还可以针对任务执行前后做一些钩子函数的实现，可以在任务执行前后做一些日志信息，这样可以多记录信息方便后面统计线程池执行任务的一些内容参数等。

2 JDK自带的构建线程池的方式

JDK中基于Executors 提供了很多线程池

2.1 FixedThreadPool

• 这个线程池的线程数是固定的，在 Executors 中第一个方法就是构建 FixedThreadPool

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

• 构建时，需要给newFixedThreadPool方法提供一个 nThreads 的属性，而这个属性其实就是当前线程池中线程的个数。当前线程池的本质其实就是使用 ThreadPoolExecutor。
• 构建好当前线程池后，线程个数已经固定好（线程是懒加载的，在构建之初，线程并没有构建出来，而是随着任务的提交才会将线程在线程池中构建出来）。如果线程没构建，线程会等着任务执行时被创建和执行；如果线程都已经构建好了，此时任务会被放到 LinkedBlockingQueue 无界队列中存放，等待线程从 LinkedBlockingQueue 中去 take 出任务，然后执行。
• 测试代码：

public static void main(String[] args) {
   
    ExecutorService threadPool = Executors.newFixedThreadPool(3);
    threadPool.execute(() -> {
   
        System.out.println("1号任务" + Thread.currentThread().getName() + System.currentTimeMillis());
        try {
   
            Thread.sleep(5000);
        } catch (InterruptedException e) {
   
            e.printStackTrace();
        }
    });
    threadPool.execute(() -> {
   
        System.out.println("2号任务" + Thread.currentThread().getName() + System.currentTimeMillis());
        try {
   
            Thread.sleep(5000);
        } catch (InterruptedException e) {
   
            e.printStackTrace();
        }
    });
    threadPool.execute(() -> {
   
        System.out.println("3号任务" + Thread.currentThread().getName() + System.currentTimeMillis());
        try {
   
            Thread.sleep(5000);
        } catch (InterruptedException e) {
   
            e.printStackTrace();
        }
    });
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

2.2 SingleThreadExecutor

• 这个线程池看名字就知道是单例线程池，线程池中只有一个工作线程在处理任务。如果业务涉及到顺序消费，可以采用 SigleThreadExecutor。

public static ExecutorService newSingleThreadExecutor() {
   
	// 在内部依然是构建了ThreadPoolExecutor，设置线程个数为1。当任务投递过来后，第一个任务会被工作线程处理，后续的任务会被扔到阻塞队列中，投递到阻塞队列中任务的顺序就是工作线程处理的顺序
	// 当前线程池可以用作顺序处理的一些业务中
    return new Executors.FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                    0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>()));
}

static class FinalizableDelegatedExecutorService extends Executors.DelegatedExecutorService {
   
  	// 线程池的使用没有区别，跟正常的ThreadPoolExecutor没区别
	FinalizableDelegatedExecutorService(ExecutorService executor) {
   
        super(executor);
    }
	// finalize是当前对象被GC干掉之前要执行的方法
	// 当前FinalizableDelegatedExecutorService的目的是为了在当前线程池被GC回收之前，可以执行shutdown，shutdown方法是将当前线程池停止，并且干掉工作线程
	// 但是不能基于这种方式保证线程池一定会执行shutdown，finalize在执行时，是守护线程，这种线程无法保证一定可以执行完毕。
	// 在使用线程池时，如果线程池是基于一个业务构建的，在使用完毕之后，一定要手动执行shutdown，否则会造成JVM中一堆线程
    protected void finalize() {
   
        super.shutdown();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

• 测试代码：

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

    threadPool.execute(() -> {
   
        System.out.println(Thread.currentThread().getName() + "," + "111");
    });
    threadPool.execute(() -> {
   
        System.out.println(Thread.currentThread().getName() + "," + "222");
    });
    threadPool.execute(() -> {
   
        System.out.println(Thread.currentThread().getName() + "," + "333");
    });
    threadPool.execute(() -> {
   
        System.out.println(Thread.currentThread().getName() + "," + "444");
    });
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

• 测试线程池使用完毕后，不执行 shutdown 的后果：
  • 如果是局部变量，仅限当前线程使用的线程池，在使用完毕之后要记得执行 shutdown，避免线程无法结束。
  • 如果是全局的线程池，很多业务都会用到，使用完毕之后不用 shutdown，因为其他业务也要执行当前线程池。

2.3 CachedThreaPool

• 看名字好像是一个缓存的线程池

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

• 当第一次提交任务到线程池时，会直接构建一个工作线程，这个工作线程执行完之后，如果60秒没有任务可以执行，那么会结束；如果等待 60 秒期间有任务进来，它会再次拿到这个任务去执行。如果后续提升任务时，没有线程是空闲的，那么久构建工作线程去执行。
• 最大的一个特点：任务只要提交到当前的 CachedThreaPool 中，就必然有工作线程可以处理。
• 测试代码：

public static void main(String[] args) {
   
    ExecutorService threadPool = Executors.newCachedThreadPool();
    for (int i = 1; i <= 200; i++) {
   
        final int x = i;
        threadPool.execute(() -> {
   
            try {
   
                Thread.sleep(1000);
            } catch (InterruptedException e) {
   
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + ":" + x);
        });
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

2.4 ScheduleThreadPool

• 看名字可以猜到当前线程池是一个定时任务的线程池，而这个线程池就是可以以一定周期去执行一个任务，或者延迟多久执行一个任务一次。

// 基于这个方法可以看到，构建的是 ScheduledThreadPoolExecutor
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
   
    return new ScheduledThreadPoolExecutor(corePoolSize);
}
public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {
   
	// ...
}
1
2
3
4
5
6
7
8
9
10
11

• 本质上还是ThreadPoolExecutor，只不是在原来的线程池基础上实现了定时任务的功能。原理是基于 DelayQueue 实现的延迟执行，周期性执行是任务执行完毕后，再次扔回到阻塞队列中。
• 测试代码：

public static void main(String[] args) {
   
    ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(10);
//        // 正常执行
//        threadPool.execute(() -> {
   
//            System.out.println(Thread.currentThread().getName() + ": 1");
//        });
//
//        // 延迟执行，执行当前任务延迟5s后执行
//        threadPool.schedule(() -> {
   
//            System.out.println(Thread.currentThread().getName() + ": 2");
//        }, 5, TimeUnit.SECONDS);
//
//        // 周期执行，当前任务第一次延迟2s执行，然后每1s执行一次
//        // 这个方法在计算下次执行时间时，是从任务刚刚开始时就计算
//        threadPool.scheduleAtFixedRate(() -> {
   
//            try {
   
//                Thread.sleep(3000);
//            } catch (InterruptedException e) {
   
//                e.printStackTrace();
//            }
//            System.out.println(Thread.currentThread().getName() + ": 3");
//        }, 2, 1, TimeUnit.SECONDS);

    threadPool.scheduleWithFixedDelay(() -> {
   
        try {
   
            Thread.sleep(3000);
        } catch (InterruptedException e) {
   
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName() + ": 4");
    }, 2, 1, TimeUnit.SECONDS);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

2.5 WorkStealingPool

• 当前JDK提供的构建线程池的方式 newWorkStealingPool 和之前的线程池有非常大的区别，之前定长、单例、缓存、定时任务都是基于 ThreadPoolExecutor 去实现的，而 WorkStealingPool 是基于 ForkJoinPool 构建的。
• ThreadPoolExecutor的核心点：只有一个阻塞队列存放当前任务
• ForkJoinPool的核心点：当有一个特别大的任务时，采用 ThreadPoolExecutor 只会有一个线程去执行，但是 ForkJoinPool 会将一个大任务拆分成很大小任务，放到当前线程的阻塞队列中，其它空闲线程就可以去处理有任务的线程的阻塞队列中的任务

• 核心就是希望没有工作线程处于空闲状态，每个线程都使用率满满。

举个例子：来一个比较大的数组，里面存满值，计算总和

• 单线程处理：

/** 非常大的数组 */
static int[] nums = new int[1_000_000_000];
// 填充值
static{
   
    for (int i = 0; i < nums.length; i++) {
   
        nums[i] = (int) ((Math.random()) * 1000);
    }
}
public static void main(String[] args) {
   
    // ===================单线程累加10亿数据================================
    System.out.println("单线程计算数组总和！");
    long start = System.nanoTime();
    int sum = 0;
    for (int num : nums) {
   
        sum += num;
    }
    long end = System.nanoTime();
    System.out.println("单线程运算结果为：" + sum + "，计算时间为：" + (end  - start));
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

• 多线程分而治之方式处理：

/**
 * 非常大的数组
 */
static int[] nums = new int[1_000_000_000];
// 填充值
static {
   
    for (int i = 0; i < nums.length; i++) {
   
        nums[i] = (int) ((Math.random()) * 1000);
    }
}
public static void main(String[] args) {
   
    // ===================单线程累加10亿数据================================
    System.out.println("单
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15