JUC包(五) 线程池

心已赠人 2022-02-27 06:30 290阅读 0赞

### 前言 ###

在前面的篇章中.我们将解了`线程/锁/多线程容器`.本章我们将介绍一个用于管理线程的容器:`线程池`.

--------------------

### 正文 ###

线程池基本构造如下所示:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly95YW54bWwuYmxvZy5jc2RuLm5ldA_size_16_color_FFFFFF_t_70]  
基本使用步骤如下:

*  步骤1: 核心线程池是否已满? 未满直接执行,满了放入等待队列`BlockedQueue`中.
 *  步骤2: 等待队列是否已满? 未满放入其中, 满了执行步骤3.
 *  步骤3: 线程池是否已满?未满放入其中, 满了执行步骤4.
 *  步骤4: 使用创建时指定的策略进行处理.  
    其关键的`execute()`部分代码如下所示:

public void execute(Runnable command) {
            if (command == null)
                throw new NullPointerException();
            int c = ctl.get();
            // 如果线程数目小于核心线程数目 创建线程完成任务
            if (workerCountOf(c) < corePoolSize) {
                if (addWorker(command, true))
                    return;
                c = ctl.get();
            }
            // 放入线程队列
            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);
            }
            // false表示非核心线程
            else if (!addWorker(command, false))
                reject(command);
        }

private boolean addWorker(Runnable firstTask, boolean core) {
            retry:
            for (;;) {
                int c = ctl.get();
                int rs = runStateOf(c);
    
                // Check if queue empty only if necessary.
                if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                       firstTask == null &&
                       ! workQueue.isEmpty()))
                    return false;
    
                for (;;) {
                    int wc = workerCountOf(c);
                    if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                        return false;
                    if (compareAndIncrementWorkerCount(c))
                        break retry;
                    c = ctl.get();  // Re-read ctl
                    if (runStateOf(c) != rs)
                        continue retry;
                    // else CAS failed due to workerCount change; retry inner loop
                }
            }
    
            boolean workerStarted = false;
            boolean workerAdded = false;
            Worker w = null;
            try {
            // this.mainLock 是 ThreadPoolExecutor的一个私有锁. 并发执行时的线程调用.
                final ReentrantLock mainLock = this.mainLock;
                w = new Worker(firstTask);
                final Thread t = w.thread;
                if (t != null) {
                    mainLock.lock();
                    try {
                        // Recheck while holding lock.
                        // Back out on ThreadFactory failure or if
                        // shut down before lock acquired.
                        int c = ctl.get();
                        int rs = runStateOf(c);
    
                        if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) {
                            if (t.isAlive()) // precheck that t is startable
                                throw new IllegalThreadStateException();
                            workers.add(w);
                            int s = workers.size();
                            if (s > largestPoolSize)
                                largestPoolSize = s;
                            workerAdded = true;
                        }
                    } finally {
                        mainLock.unlock();
                    }
                    if (workerAdded) {
                    	// 启动Worker线程 Woker线程会调用run()方法执行
                        t.start();
                        workerStarted = true;
                    }
                }
            } finally {
                if (! workerStarted)
                    addWorkerFailed(w);
            }
            return workerStarted;
        }

在`addWorker()`方法后, 会自动调用Worker的`run()`方法进行运行.

# Worker类的run方法如下
    public void run() {
         runWorker(this);
    }
    final void runWorker(Worker w) {
            Thread wt = Thread.currentThread();
            Runnable task = w.firstTask;
            w.firstTask = null;
            w.unlock(); // allow interrupts
            boolean completedAbruptly = true;
            try {
                while (task != null || (task = getTask()) != null) {
                    w.lock();
                    // If pool is stopping, ensure thread is interrupted;
                    // if not, ensure thread is not interrupted.  This
                    // requires a recheck in second case to deal with
                    // shutdownNow race while clearing interrupt
                    if ((runStateAtLeast(ctl.get(), STOP) ||
                         (Thread.interrupted() &&
                          runStateAtLeast(ctl.get(), STOP))) &&
                        !wt.isInterrupted())
                        wt.interrupt();
                    try {
                        beforeExecute(wt, task);
                        Throwable thrown = null;
                        try {
                            task.run();
                        } catch (RuntimeException x) {
                            thrown = x; throw x;
                        } catch (Error x) {
                            thrown = x; throw x;
                        } catch (Throwable x) {
                            thrown = x; throw new Error(x);
                        } finally {
                            afterExecute(task, thrown);
                        }
                    } finally {
                        task = null;
                        w.completedTasks++;
                        w.unlock();
                    }
                }
                completedAbruptly = false;
            } finally {
                processWorkerExit(w, completedAbruptly);
            }
        }

--------------------

### 基本使用 ###

*  初始化  
    `new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue)`  
    `corePoolSize`: 核心线程池线程数目  
    `maximumPoolSize`: 最大线程数目  
    `keepAliveTime`: 非核心线程的存活时间  
    `unit`: 非核心线程的存活时间单位  
    `workQueue`: 阻塞队列类型.(`ArrayBlockingQueue`/`LinkedBlockingQueue`/`SynchronousQueue`/`PriorityBlockingQueue`)  
    `ThreadFactory`: 设置线程的创建工厂.例如,线程起名工厂`new ThreadFactoryBuilder().setNameFormat("xx-xxx-%d").build()`  
    `RejectedExecutionHandler`饱和策略: `AbortPolicy(抛出异常)`/`CallerRunsPolicy(调用者所在线程完成任务)` / `DiscardOldestPolicy(丢弃最近任务)` / `DiscardPolicy(不处理,直接丢弃)`
 *  提交任务  
    `execute()`: 无返回值  
    `submit()`: 返回值`Future`
 *  关闭线程池 `threadPool.shutdown();`
 *  总览

/**
     * 9-1 测试线程池的基本使用功能.
     * 
     * */
    public class ThreadPoolExecutorTest {
    	public static void main(String[] args) throws InterruptedException {
    		ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 100, 1000, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
    		
    		threadPool.submit(new Runnable(){
    
    			@Override
    			public void run() {
    				while(true){
    					System.out.println(Thread.currentThread().getName()+":"+System.currentTimeMillis());
    				}
    			}
    			
    		});
    		Thread.sleep(10000);
    		threadPool.shutdown();
    	}
    }

--------------------

### Others ###

\[1\]. [线程池之ThreadPoolExecutor状态控制][ThreadPoolExecutor]  
\[2\]. [线程池的优雅关闭实践][Link 1]  
\[3\]. [ThreadPoolExecutor的clt：runState&workCount][ThreadPoolExecutor_clt_runState_workCount]

--------------------

### Tips 2019-04-30 ###

隔了一个月,再看文章. 确实有很多不当之处.

当一个人提问和接触Java线程池的知识的时候, 应对是安装如下的顺序:

1.  线程池如何声明和使用
2.  线程池声明的各个参数
3.  线程池开启和关闭的方式
4.  线程池使用的流程
5.  线程池的内部原理

> Q1&Q2: 线程池的声明  
> 线程池通过`new ThreadPoolExecutor(xx,xx)`进行创建. 其中包括多个参数, `coreSize`核心线程池数目, `maxSize`最大线程池数目, `keepaliveTime`空闲Worker线程存活时间, `TimeUnit`存活时间单位, `BlockQueue`消息队列类型, `ThreadFactory`线程创建工厂, 线程溢出策略`Pliocy`.

> Q1: 线程池的使用  
> 线程池的使用主要有两种方法: 1. pool.submit(new Runnable()) 2. pool.execute(new Runnable()). 两种方式的区别在于, `submit`方法有返回值, `execute`方法没有返回值.

> Q3: 线程池的开启和关闭  
> 线程池开启, 不需要特殊的方法. 通过`submit`/`execute`提交后, 线程即会启动.  
> 线程池关闭, 使用`shutdown()`方法关闭线程池. 不关闭, 会导致线程池内的某些线程会一直运行.  
> \[1\]. [Java线程池关闭2-shutdown和awaitTermination][Java_2-shutdown_awaitTermination]  
> \[2\]. [线程池的优雅关闭实践][Link 1]

> Q4: 线程池使用流程  
> 线程池使用流程:
> 
> 1.  初始化: 通过new ThreadPoolExecutor(xxx,xxx)声明线程池.
> 2.  任务提交: 当第一个任务`Runnable()`进行提交后, 会运行execute()方法.
> 3.  任务调度&执行:  
>     3-1. 首先会判断当前核心线程数目, 如果小于.则创建核心线程执行.  
>     3-2. 如果核心线程数目已满, 会将任务放入`BlockingQueue`队列.等待后续执行.  
>     3-3. 如果前面两者都满了, 会将其声明一个非核心线程执行. 当线程总数大于`maxSize`, 会根据处理策略`Pliocy`进行处理. 最简单的策略就是抛出异常.  
>     Tips: 不论`submit`还是`execute`, 最后都是调用`execute`方法.
> 
>     # abstract class AbstractExecutorService implements ExecutorService {
>     
>         public Future<?> submit(Runnable task) {
>           if (task == null) throw new NullPointerException();
>            RunnableFuture<Void> ftask = newTaskFor(task, null);
>            execute(ftask);
>            return ftask;
>        }
>       }

> Q5: 线程池的核心原理  
> 线程池核心原理, 其实就是线程池使用流程. 同上.

> Q6: 默认的线程声明工厂

# step1 
      public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue) {
            this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                 Executors.defaultThreadFactory(), defaultHandler);
        }
      # step2      
        public static ThreadFactory defaultThreadFactory() {
            return new DefaultThreadFactory();
        }
    
    # step3   
    static class DefaultThreadFactory implements ThreadFactory {
            private static final AtomicInteger poolNumber = new AtomicInteger(1);
            private final ThreadGroup group;
            private final AtomicInteger threadNumber = new AtomicInteger(1);
            private final String namePrefix;
    
            DefaultThreadFactory() {
                SecurityManager s = System.getSecurityManager();
                group = (s != null) ? s.getThreadGroup() :
                                      Thread.currentThread().getThreadGroup();
                namePrefix = "pool-" +
                              poolNumber.getAndIncrement() +
                             "-thread-";
            }
    
            public Thread newThread(Runnable r) {
                Thread t = new Thread(group, r,
                                      namePrefix + threadNumber.getAndIncrement(),
                                      0);
                if (t.isDaemon())
                    t.setDaemon(false);
                if (t.getPriority() != Thread.NORM_PRIORITY)
                    t.setPriority(Thread.NORM_PRIORITY);
                return t;
            }
        }

--------------------

### Reference ###

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly95YW54bWwuYmxvZy5jc2RuLm5ldA_size_16_color_FFFFFF_t_70]: /images/20220227/f5211c06e45f402680666bca84c81617.png
[ThreadPoolExecutor]: https://www.jianshu.com/p/18065a78178b
[Link 1]: https://www.jianshu.com/p/bdf06e2c1541
[ThreadPoolExecutor_clt_runState_workCount]: https://blog.csdn.net/luzhen1012/article/details/79230091
[Java_2-shutdown_awaitTermination]: https://blog.csdn.net/woshixuye/article/details/52754063