「Java程序设计」-多线程|线程池

      如果并发的请求数量非常多，但每个线程执行的时间很短，这样就会频繁的创建和销毁线程，如此一来会大大降低系统的效率。可能出现服务器在为每个请求创建新线程和销毁线程上花费的时间和消耗的系统资源要比处理实际的用户请求的时间和资源更多。所以应该使用线程池处理类似问题。
使用线程池的场景：

  1. 单个任务处理时间比较短
  2. 需要处理任务的数量很大

使用线程池的好处：

  1. 降低资源消耗：通过复用已存在的线程和降低线程关闭的次数来尽可能降低系统性能损耗
  2. 提升系统响应速度：通过复用线程，省去创建线程的过程，因此整体上提升了系统的响应速度
  3. 提高线程的可管理性：线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性

一、线程池创建

当一个并发任务提交给线程池，线程池分配线程去执行任务的过程如下图：

线程池分配线程流程

从图可以看出，线程池执行所提交的任务过程主要有这样几个阶段：

  1. 先判断线程池中核心线程池所有的线程是否都在执行任务。如果不是，则新创建一个线程执行刚提交的任务，否则，核心线程池中所有的线程都在执行任务，则进入第2步；
  2. 判断当前阻塞队列是否已满，如果未满，则将提交的任务放置在阻塞队列中；否则，则进入第3步；
  3. 判断线程池中所有的线程是否都在执行任务，如果没有，则创建一个新的线程来执行任务，否则，则交给饱和策略进行处理

线程池主要使用ThreadPoolExecutor类来完成，参数最多的构造方法如下：

1
2
3
4
5
6
7
public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler)

  • corePoolSize
        表示核心线程池的大小，当提交一个任务时，如果当前核心线程池的线程个数没有达到corePoolSize，则会创建新的线程来执行所提交的任务，即使当前核心线程池有空闲的线程。如果当前核心线程池的线程个数已经达到了corePoolSize，则不再重新创建线程。
  • maximumPoolSize
        表示线程池能创建线程的最大个数。如果当阻塞队列已满时，并且当前线程池线程个数没有超过maximumPoolSize的话，就会创建新的线程来执行任务。
  • keepAliveTime
        空闲线程存活时间。如果当前线程池的线程个数已经超过了corePoolSize，并且线程空闲时间超过了keepAliveTime的话，就会将这些空闲线程销毁，这样可以尽可能降低系统资源消耗。
  • unit
        时间单位。为keepAliveTime指定时间单位。
  • workQueue
        阻塞队列。用于保存任务的阻塞队列。常用的如LinkedBlockingQueue
  • threadFactory
        创建线程的工厂类。可以通过指定线程工厂为每个创建出来的线程设置更有意义的名字，方便问题的排查。
  • handler
        饱和策略。当线程池的阻塞队列已满和指定的线程都已经开启，说明当前线程池已经处于饱和状态了，那么就需要采用一种策略来处理这种情况。采用的策略有这几种：
        a. AbortPolicy： 直接拒绝所提交的任务，并抛出RejectedExecutionException异常；
        b. CallerRunsPolicy：只用调用者所在的线程来执行任务；
        c. DiscardPolicy：不处理直接丢弃掉任务；
        d. DiscardOldestPolicy：丢弃掉阻塞队列中存放时间最久的任务，执行当前任务

二、线程池执行逻辑

通过查看execute()方法的源码：

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
42
43
44
public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    /*
     * Proceed in 3 steps:
     *
     * 1. If fewer than corePoolSize threads are running, try to
     * start a new thread with the given command as its first
     * task.  The call to addWorker atomically checks runState and
     * workerCount, and so prevents false alarms that would add
     * threads when it shouldn't, by returning false.
     *
     * 2. If a task can be successfully queued, then we still need
     * to double-check whether we should have added a thread
     * (because existing ones died since last checking) or that
     * the pool shut down since entry into this method. So we
     * recheck state and if necessary roll back the enqueuing if
     * stopped, or start a new thread if there are none.
     *
     * 3. If we cannot queue task, then we try to add a new
     * thread.  If it fails, we know we are shut down or saturated
     * and so reject the task.
     */
    int c = ctl.get();
    //如果线程池的线程个数<corePoolSize，则创建新线程执行当前任务
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    
    //如果线程个数>corePoolSize或者创建线程失败，则将任务存放在阻塞队列workQueue中
    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);
}

ThreadPoolExecutor的execute方法执行逻辑见下图：

execute执行逻辑图
  1. 如果当前运行的线程少于corePoolSize，则会创建新的线程来执行新的任务；
  2. 如果运行的线程个数等于或者大于corePoolSize，则会将提交的任务存放到阻塞队列workQueue中；
  3. 如果当前workQueue队列已满的话，则会创建新的线程来执行任务；
  4. 如果线程个数已经超过了maximumPoolSize，则会使用饱和策略RejectedExecutionHandler来进行处理。
    需要注意的是，线程池的设计思想就是使用了核心线程池corePoolSize，阻塞队列workQueue和线程池maximumPoolSize，这样的缓存策略来处理任务，实际上这样的设计思想在需要框架中都会使用。

三、线程池的关闭

      关闭线程池，可以使用shutdown和shutdownNow这两个方法，它们都会遍历线程池中的所有线程，然后依次中断线程。区别如下：shutdown方法会将正在执行的任务继续执行完，而shutdownNow会直接中断正在执行的任务。调用了这两个方法的任意一个，isShutdown方法都会返回true，当所有的线程都关闭成功，才表示线程池成功关闭，这时调用isTerminated方法才会返回true。

四、合理配置线程池参数

想要合理的配置线程池，就必须首先分析任务特性：

  1. 任务的性质：CPU密集型任务/IO密集型任务
  2. 任务的优先级
  3. 任务的执行时间
  4. 任务的依赖性：是否依赖其他系统资源，如数据库连接

      CPU密集型任务（计算型任务）配置尽可能少的线程数量，如配置Ncpu+1个线程的线程池。IO密集型任务则由于需要等待IO操作，线程并不是一直在执行任务，则配置尽可能多的线程，如2xNcpu。
      优先级不同的任务可以使用优先级队列PriorityBlockingQueue来处理。它可以让优先级高的任务先得到执行，需要注意的是如果一直有优先级高的任务提交到队列里，那么优先级低的任务可能永远不能执行。
      执行时间不同的任务可以交给不同规模的线程池来处理，或者也可以使用优先级队列，让执行时间短的任务先执行。
      依赖数据库连接池的任务，因为线程提交SQL后需要等待数据库返回结果，如果等待的时间越长CPU空闲时间就越长，那么线程数应该设置越大，这样才能更好的利用CPU。
      最后，阻塞队列最好是使用有界队列，如果采用无界队列的话，一旦任务积压在阻塞队列中的话就会占用过多的内存资源，甚至会使得系统崩溃。


本文参考资料：https://github.com/CL0610/Java-concurrency
赏

今晚终于可以加个鸡腿了~

支付宝打赏
微信打赏