从ThreadPoolExecutor看线程池

矫情吗；* 2022-12-20 06:12 262阅读 0赞

首先复习下创建线程的几种方式

  1. 1、实现runnable接口
  3.     new Thread(() -> log.info("方式一：实现runnable接口")).start();
  5. 2、实现callable接口
  7.     FutureTask<String> task = new FutureTask<>(() -> "方式二：实现callable接口");
  8.     new Thread(task).start();
  10. 3、继承Thread类
  12.     class MyThread extends Thread{ 
  13.         @Override
  14.         public void run() { 
  15.             log.info("方式三：继承thread类");
  16.         }
  17.     };
  19. 4、使用线程池  这个只是创建了线程池，但是线程具体需要执行的任务还是需要在方法中去指定
  21.     ExecutorService executorService = Executors.newCachedThreadPool();
  22.     executorService.submit(() -> log.info("方式四：使用线程池"));

在这里插入图片描述
ok 现在我们开始来熟悉线程池

1、创建线程池

  1. public ThreadPoolExecutor(int corePoolSize, //核心线程数
  2.                           int maximumPoolSize, //最大线程数
  3.                           long keepAliveTime, //存活时间
  4.                           TimeUnit unit, //时间单位
  5.                           BlockingQueue<Runnable> workQueue, //阻塞队列，用来储存线程
  6.                           ThreadFactory threadFactory, //创建线程的工厂
  7.                           RejectedExecutionHandler handler //拒绝策略 ) { 
  8. }
  10. //具体代码引发的问题
  12. public class ThreadPoolListener implements BeanPostProcessor , InitializingBean { 
  13.     //设置阻塞队列大小为2
  14.     BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(2);
  15.     //设置核心线程数为4 最大线程数为6
  16.     ThreadPoolExecutor pool = new ThreadPoolExecutor(4, 6, 10, TimeUnit.SECONDS, queue, (runnable, executor) -> { 
  17.         //拒绝策略需要处理的业务逻辑
  18.         log.info("do something ~");
  19.         log.info(runnable.toString());
  20.         log.info(executor.toString());
  21.     });
  22.     //定时任务处理业务~~~
  23.     @Scheduled(cron = "*/5 * * * * *")
  24.     @Async
  25.     public void logCollection(){ 
  26.         //打印线程池的工作中的线程数以及线程id 队列的大小
  27.         pool.execute(() ->{ 
  28.             log.info("测试线程池工作原理 线程数:{},当前线程id:{}",pool.getPoolSize(),Thread.currentThread().getId());
  29.             log.info(queue.size()+"----->>队列大小");
  30.             if(queue.size()>0){ 
  31.                 log.info("队列中的数据:{}",queue.peek());
  32.             }
  33.             try { 
  34.                 Thread.sleep(100000);
  35.             } catch (InterruptedException e) { 
  36.                 e.printStackTrace();
  37.             }
  38.         });
  39.     }
  40.     @Override
  41.     public void afterPropertiesSet() throws Exception { 
  42.         log.info("属性设置完毕 队列容量:{}",queue.size());
  43.     }
  44. }

在这里插入图片描述
从图中可以看到，在线程数量超过了核心线程之后，队列立马就满了 这是为什么?

从execute入手，看看他都做了什么事情

  1. public void execute(Runnable command) { 
  2.     if (command == null)
  3.         throw new NullPointerException();
  4.     //ctl:原子计数器，来保存当前的正在执行的线程数以及线程状态（高3位存状态，低29位存线程数）
  5.     int c = ctl.get();
  6.     //执行中的线程数是否小于核心线程数
  7.     if (workerCountOf(c) < corePoolSize) { 
  8.         //添加一个worker来处理任务
  9.         if (addWorker(command, true))
  10.             return;
  11.         //走到这里说明worker添加失败了 再次获取下线程的状态数
  12.         c = ctl.get();
  13.     }
  14.     //如果线程池是运行的状态，将本任务加入到队列中
  15.     if (isRunning(c) && workQueue.offer(command)) { 
  16.         int recheck = ctl.get();
  17.         //再判断线程池是不是运行中 不是的话，将本线程移出队列
  18.         if (! isRunning(recheck) && remove(command))
  19.             // 然后执行拒绝策略
  20.             reject(command);
  21.         //线程数量为0的话， 开启创建一个新的worker 注意传的参数
  22.         else if (workerCountOf(recheck) == 0)
  23.             //开启一个新worker，去处理队列中的任务 这里false，在方法里面会作为判断允许线程的数量的条件
  24.             //上面一行话看不懂，可以去看看这个方法，就自然懂了
  25.             addWorker(null, false);
  26.     }
  27.     //线程池不是运行中 或者任务加入队列失败 再次尝试添加一个worker去处理任务
  28.     else if (!addWorker(command, false))
  29.         //添加失败，执行拒绝策略
  30.         reject(command);
  31. }

从上面我们可以看到，在判断完线程池的状态以及数量之后，要么执行拒绝策略，要么创建一个worker，要么将任务加入到队列中，所以具体是怎么去处理线程任务？我们来分析addWorker

  1. private boolean addWorker(Runnable firstTask, boolean core) { 
  2.   retry:
  3.     for (;;) { 
  4.         int c = ctl.get();
  5.         int rs = runStateOf(c);
  7.        //如果线程已经关闭 并且 任务为null，队列不为空 直接返回 
  8.        //（因为线程已经关闭，任务队列也为空，当前也没有需要执行的任务。所以直接返回）
  9.        //(如果任务队列不为空，或者当前需要执行的任务不为null，并且线程池状态为shutdown，
  10.        //线程池会执行完当前的任务，如果状态为stop，则直接返回，所有任务都会被抛弃)
  11.         if (rs >= SHUTDOWN && 
  12.             ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()))
  13.             return false;
  15.         for (;;) { 
  16.             int wc = workerCountOf(c);
  17.             //当前线程数量线程最大允许数量 直接返回 
  18.             if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
  19.                 return false;
  20.             //尝试增加线程数量
  21.             if (compareAndIncrementWorkerCount(c))
  22.                 //成功的话，跳出外层循环
  23.                 break retry;
  24.             //走到这里说明由于并发导致线程数量增加失败
  25.             c = ctl.get();  
  26.             //再次判断下线程池状态 如果状态被改变了，那么执行下一次的外层循环 在去获取线程池状态、数量等信息
  27.             if (runStateOf(c) != rs)
  28.                 continue retry;
  29.         }
  30.     }
  32.     //到了这里，说明已经成功增加了线程数量，才会跳出上面的循环
  33.     boolean workerStarted = false;
  34.     boolean workerAdded = false;
  35.     Worker w = null;
  36.     try { 
  37.         //增加一个新的worker，将我们传过来的任务传进去，作为线程的第一个任务去执行
  38.         //worker中会创建一个线程，一个worker对应一个线程
  39.         w = new Worker(firstTask);
  40.         //拿到worker中的线程
  41.         final Thread t = w.thread;
  42.         if (t != null) { 
  43.             final ReentrantLock mainLock = this.mainLock;
  44.             //加独占锁 这一块就与aqs关联起来了 怎么加锁，查看我前面的文章
  45.             mainLock.lock();
  46.             try { 
  47.                 int rs = runStateOf(ctl.get());
  48.                 //线程池是运行中的状态 或者是shutdown，并且本任务不为null
  49.                 if (rs < SHUTDOWN ||
  50.                     (rs == SHUTDOWN && firstTask == null)) { 
  51.                     //本线程已经在运转，直接抛异常
  52.                     if (t.isAlive()) // precheck that t is startable
  53.                         throw new IllegalThreadStateException();
  54.                     //将这个worker加入到工作组中
  55.                     workers.add(w);
  56.                     int s = workers.size();
  57.                     if (s > largestPoolSize)
  58.                         largestPoolSize = s;
  59.                     workerAdded = true;
  60.                 }
  61.             } finally { 
  62.                 //释放锁 看到这里，我们可以知道，
  63.                 //上面枷锁，只是为了保证在添加woker到工作组中，线程池的状态不被改变
  64.                 mainLock.unlock();
  65.             }
  66.             //worker被成功加入到工作组中了。再来启动这个worker去工作
  67.             if (workerAdded) { 
  68.                 //重点！！！
  69.                 t.start();
  70.                 workerStarted = true;
  71.             }
  72.         }
  73.     } finally { 
  74.         if (! workerStarted)
  75.             //worker没启动，需要做一些清理工作，如前面 workCount 加了 1，将其减掉
  76.             addWorkerFailed(w);
  77.     }
  78.     //返回worker的工作状态
  79.     return workerStarted;
  80. }

我们都知道，在调用了thread.start()方法之后，他会启动线程，然后去执行线程的run方法。同样，我们在将worker启动之后，也会调用他的run方法。我们来看run里面都做了什么

这里做一个简单说明：在创建worker的时候，他们将我们需要执行的任务作为必要的构造器参数传进去，在构造方法里面，会对他做一个包装之后，在赋给worker中的thread字段。

  1. public void run() { 
  2.     runWorker(this);
  3. }
  5. final void runWorker(Worker w) { 
  6.     Thread wt = Thread.currentThread();
  7.     //获取第一个任务
  8.     Runnable task = w.firstTask;
  9.     //将第一个任务置空
  10.     w.firstTask = null;
  11.     w.unlock(); // allow interrupts
  12.     boolean completedAbruptly = true;
  13.     try { 
  14.         //第一个任务不为null 或者为null，那么就去取任务
  15.         while (task != null || (task = getTask()) != null) { 
  16.             //加锁
  17.             w.lock();
  18.             //线程池状态为stop，那么本线程直接中断进行
  19.             if ((runStateAtLeast(ctl.get(), STOP) ||
  20.                  (Thread.interrupted() &&
  21.                   runStateAtLeast(ctl.get(), STOP))) &&
  22.                 !wt.isInterrupted())
  23.                 wt.interrupt();
  24.             try { 
  25.                 //执行之前的处理 这一块为空方法 由此可见我们可以继承然后重写这个方法
  26.                 beforeExecute(wt, task);
  27.                 Throwable thrown = null;
  28.                 try { 
  29.                     //执行我们的业务
  30.                     task.run();
  31.                 } catch (RuntimeException x) { 
  32.                     thrown = x; throw x;
  33.                 } catch (Error x) { 
  34.                     thrown = x; throw x;
  35.                 } catch (Throwable x) { 
  36.                     thrown = x; throw new Error(x);
  37.                 } finally { 
  38.                     //执行之后的处理
  39.                     afterExecute(task, thrown);
  40.                 }
  41.             } finally { 
  42.                 //任务置空 那么就可以获取下一个任务
  43.                 task = null;
  44.                 //完成的任务数
  45.                 w.completedTasks++;
  46.                 //释放锁
  47.                 w.unlock();
  48.             }
  49.         }
  50.         completedAbruptly = false;
  51.     } finally { 
  52.         //到这里有两种情况
  53.         //1 任务为空 那么正常走到这里completedAbruptly 已被改为false
  54.         //2 出现了异常 此时completedAbruptly由于异常，所以他的值还是true
  55.         processWorkerExit(w, completedAbruptly);
  56.     }
  57. }

任务为空 或者出现异常之后，来在这里做处理

  1. private void processWorkerExit(Worker w, boolean completedAbruptly) { 
  2.     //如果有异常，那么减少线程的数量
  3.     if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
  4.         decrementWorkerCount();
  6.     final ReentrantLock mainLock = this.mainLock;
  7.     mainLock.lock();
  8.     try { 
  9.         completedTaskCount += w.completedTasks;
  10.         //将本worker移除工作组
  11.         workers.remove(w);
  12.     } finally { 
  13.         mainLock.unlock();
  14.     }
  15.     //更改线程池状态 略过吧 不是重点 感兴趣的可以自己看 也很简单 反正我没看
  16.     tryTerminate();
  18.     int c = ctl.get();
  19.     //如果线程池的状态<stop
  20.     if (runStateLessThan(c, STOP)) { 
  21.         //没出现异常
  22.         if (!completedAbruptly) { 
  23.             int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
  24.             //队列不为空，则保证最少有一个线程存在，来处理这个队列
  25.             if (min == 0 && ! workQueue.isEmpty())
  26.                 min = 1;
  27.             if (workerCountOf(c) >= min)
  28.                 return; // replacement not needed
  29.         }
  30.         addWorker(null, false);
  31.     }
  32. }

再来看看 线程是如何获取任务的

  1. private Runnable getTask() { 
  2.     boolean timedOut = false; // Did the last poll() time out?
  4.     for (;;) { 
  5.         int c = ctl.get();
  6.         int rs = runStateOf(c);
  7.         //线程池状态>=stop 或者线程池状态=shutdown，队列为空 也是再次保证，线程shutdown之后，不在接收新的创建线程然后去处理任务的请求 但是已经存在的任务还是可以正常去处理
  8.         if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { 
  9.             //减少线程的数量
  10.             decrementWorkerCount();
  11.             return null;
  12.         }
  14.         int wc = workerCountOf(c);
  15.         //正在工作的线程是否超过核心线程数
  16.         boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
  17.         //队列为空 或者线程池状态改变 那么也返回null
  18.         if ((wc > maximumPoolSize || (timed && timedOut))
  19.             && (wc > 1 || workQueue.isEmpty())) { 
  20.             if (compareAndDecrementWorkerCount(c))
  21.                 return null;
  22.             continue;
  23.         }
  25.         try { 
  26.             //获取队列中的任务
  27.             Runnable r = timed ?
  28.                 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
  29.                 workQueue.take();
  30.             if (r != null)
  31.                 return r;
  32.             timedOut = true;
  33.         } catch (InterruptedException retry) { 
  34.             // 如果此 worker 发生了中断，采取的方案是重试
  35.             // 解释下为什么会发生中断，这个读者要去看 setMaximumPoolSize 方法。
  37.             // 如果开发者将 maximumPoolSize 调小了，导致其小于当前的 workers 数量，
  38.             // 那么意味着超出的部分线程要被关闭。重新进入 for 循环，自然会有部分线程会返回 null
  39.             timedOut = false;
  40.         }
  41.     }
  42. }

至此，关于任务是怎么加入到队列，以及又是怎么被取出来然后去执行的已经很清晰了

其他一些细节点
corePoolSize

核心线程数，不要抠字眼，反正先记着有这么个属性就可以了。

maximumPoolSize

最大线程数，线程池允许创建的最大线程数。

workQueue

任务队列，BlockingQueue 接口的某个实现（常使用 ArrayBlockingQueue 和 LinkedBlockingQueue）。

keepAliveTime

空闲线程的保活时间，如果某线程的空闲时间超过这个值都没有任务给它做，那么可以被关闭了。注意这个值并不会对所有线程起作用，如果线程池中的线程数少于等于核心线程数 corePoolSize，那么这些线程不会因为空闲太长时间而被关闭，当然，也可以通过调用 allowCoreThreadTimeOut(true)使核心线程数内的线程也可以被回收。

threadFactory

用于生成线程，一般我们可以用默认的就可以了。通常，我们可以通过它将我们的线程的名字设置得比较可读一些，如 Message-Thread-1， Message-Thread-2 类似这样。

handler：

当线程池已经满了，但是又有新的任务提交的时候，该采取什么策略由这个来指定。有几种方式可供选择，像抛出异常、直接拒绝然后返回等，也可以自己实现相应的接口实现自己的逻辑，这个之后再说。

参考文章：https://javadoop.com/post/java-thread-pool\#toc\_4

发表评论

表情：
评论列表 （有 0 条评论，262人围观）

还没有评论，来说两句吧...

相关阅读