您的位置 首页 java

Java高并发之线程池详解

智云科技发布于 2023年7月3日评论关闭阅读(65)

Java高并发之线程池详解

（一）为什么要用线程池

  • ① 线程是不是越多越好

1.不一定是越多越好，首先线程在java里面是一个对象，只要是对象肯定要占用一定的资源，更多的是操作系统保护的资源，线程的创建和销毁都需要时间和空间，如果线程的（创建时间+销毁时间）大于执行时间 就很不划算了。举个例子，创建时间1秒，销毁时间1秒，任务执行也就是1秒，这就没必要搞一个新的线程了，就考虑线程能否复用。

2.java对象占用堆内存，操作系统的线程也会占用内存，对象本身也会占用堆内存，根据JVM规范一个线程最大的栈是1M，达到最大了，栈空间就需要去系统内存中进行分配的，线程不管是占用了多少，只要线程一多，会消耗更多的内存。

3.多线程下操作系统在处理的时候，CPU时间片的增强就会有一个频繁的切换系统上下文，每个线程都想被运行，导致每个线程都执行的很慢，不能专心执行某一个线程。

线程池的推出，就是为了方便的控制 线程数 量，一般生产环境都会用到线程池。

（二）线程池的原理

  1. 线程池管理：用于创建并管理线程池，包括创建线程池，销毁线程池，添加新任务。
  2. 工作线程：线程池中线程，在没有任务时处于等待状态，可以循环的执行任务。
  3. 任务接口：如何交给任务执行，每个任务必须事先的接口，以供工作线程调度任务的执行，它主要规定了任务的入口，任务执行完后的收尾工作，任务的执行状态等。
  4. 任务队列：用户存放没有处理的任务，提供一种缓冲机制。

Java高并发之线程池详解

（三）线程池接口定义和实现类

Java高并发之线程池详解

（四）ExeutorService API

Java高并发之线程池详解

scheduleAtFixedRate 一次任务执行市场超过了周期时间，下一次任务会等到该次任务执行结束后，立刻执行，这也是它和scheduleWithFixedDelay 的重要区别

scheduleWithFixedDelay 一次任务执行市场超过了周期时间，下一次任务会等到该次任务执行结束后，需要等待延时时间后，才执行。

（五）Executors工具类

  1. newFixedThreadPool(int nThreads)创建一个固定大小、任务队列容量无界的线程池。核心线程数=最大线程数。
  2. newCachedThreadPool（）创建的是一个大小无界的缓冲线程池。他的任务队列是一个同步队列。任务加入到池中，如果池中有空闲线程，则用空闲线程执行，如无则创建新线程执行。池中的线程空闲超过60秒，将被销毁释放。线程数随任务的多是变化。适用于执行耗时较小的异步任务。池的核心线程数=0，最大线程数=Integer.MAX_VALUE
  3. newSingleThreadExecutor() 只有一个线程来执行无界任务队列的单一线程池。该线程池确保任务按加入的顺序一个一个依次执行。当唯一的线程因任务异常终止时，将创建一个新的线程来执行后续的任务。与newFixedThreadPool(1)的区别在于，单一线程池的池大小在newSingleThreadExecutor方法中硬编码，不能在改变。
  4. newScheduledThreadPool（int corePoolSize）能定时执行任务的线程池。该池的核心线程数由参数指定，最大线程数=Integer.MAX_VALUE

（六） 线程池执行原理分析

  1. 是否达到核心线程数量？没达到，创建一个工作线程来执行任务。
  2. 工作队列是否已满？没满，则将新提交的任务存储在任务队列中。
  3. 是否达到线程池最大数量？没达到，则创建一个新的工作线程来执行任务。
  4. 最后，执行拒绝策略来处理这个任务。

Java高并发之线程池详解

（七）源码 

import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/** 线程池的使用 */public class Demo9 {
 /**
 * 测试： 提交15个执行时间需要3秒的任务,看线程池的状况
 *
 * @param threadPoolExecutor 传入不同的线程池，看不同的结果
 * @throws Exception
 */ public void testCommon(ThreadPoolExecutor threadPoolExecutor) throws Exception {
 // 测试： 提交15个执行时间需要3秒的任务，看超过大小的2个，对应的处理情况
 for (int i = 0; i < 15; i++) {
 int n = i;
 threadPoolExecutor.submit(new Runnable() {
 public void run() {
 try {
 System.out.println("开始执行：" + n);
 Thread.sleep(3000L);
 System.err.println("执行结束:" + n);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 }
 });
 System.out.println("任务提交成功 :" + i);
 }
 // 查看线程数量，查看队列等待数量
 Thread.sleep(500L);
 System.out.println("当前线程池线程数量为：" + threadPoolExecutor.getPoolSize());
 System.out.println("当前线程池等待的数量为：" + threadPoolExecutor.getQueue().size());
 // 等待15秒，查看线程数量和队列数量（理论上，会被超出核心线程数量的线程自动销毁）
 Thread.sleep(15000L);
 System.out.println("当前线程池线程数量为：" + threadPoolExecutor.getPoolSize());
 System.out.println("当前线程池等待的数量为：" + threadPoolExecutor.getQueue().size());
 }
 /**
 * 1、线程池信息： 核心线程数量5，最大数量10，无界队列，超出核心线程数量的线程存活时间：5秒， 指定拒绝策略的
 *
 * @throws Exception
 */ private void threadPoolExecutorTest1() throws Exception {
 ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS,
 new LinkedBlockingQueue<Runnable>());
 testCommon(threadPoolExecutor);
 // 预计结果：线程池线程数量为：5,超出数量的任务，其他的进入队列中等待被执行
 }
 /**
 * 2、 线程池信息： 核心线程数量5，最大数量10，队列大小3，超出核心线程数量的线程存活时间：5秒， 指定拒绝策略的
 *
 * @throws Exception
 */ private void threadPoolExecutorTest2() throws Exception {
 // 创建一个 核心线程数量为5，最大数量为10,等待队列最大是3 的线程池，也就是最大容纳13个任务。
 // 默认的策略是抛出RejectedExecutionException异常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
 ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS,
 new LinkedBlockingQueue<Runnable>(3), new RejectedExecutionHandler() {
 public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
 System.err.println("有任务被拒绝执行了");
 }
 });
 testCommon(threadPoolExecutor);
 // 预计结果：
 // 1、 5个任务直接分配线程开始执行
 // 2、 3个任务进入等待队列
 // 3、 队列不够用，临时加开5个线程来执行任务(5秒没活干就销毁)
 // 4、 队列和线程池都满了，剩下2个任务，没资源了，被拒绝执行。
 // 5、 任务执行，5秒后，如果无任务可执行，销毁临时创建的5个线程
 }
 /**
 * 3、 线程池信息： 核心线程数量5，最大数量5，无界队列，超出核心线程数量的线程存活时间：5秒
 *
 * @throws Exception
 */ private void threadPoolExecutorTest3() throws Exception {
 // 和Executors.newFixedThreadPool(int nThreads)一样的
 ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS,
 new LinkedBlockingQueue<Runnable>());
 testCommon(threadPoolExecutor);
 // 预计结：线程池线程数量为：5，超出数量的任务，其他的进入队列中等待被执行
 }
 /**
 * 4、 线程池信息：
 * 核心线程数量0，最大数量Integer.MAX_VALUE，SynchronousQueue队列，超出核心线程数量的线程存活时间：60秒
 *
 * @throws Exception
 */ private void threadPoolExecutorTest4() throws Exception {
 // SynchronousQueue，实际上它不是一个真正的队列，因为它不会为队列中元素维护存储空间。与其他队列不同的是，它维护一组线程，这些线程在等待着把元素加入或移出队列。
 // 在使用SynchronousQueue作为工作队列的前提下，客户端代码向线程池提交任务时，
 // 而线程池中又没有空闲的线程能够从SynchronousQueue队列实例中取一个任务，
 // 那么相应的offer方法调用就会失败（即任务没有被存入工作队列）。
 // 此时，ThreadPoolExecutor会新建一个新的工作者线程用于对这个入队列失败的任务进行处理（假设此时线程池的大小还未达到其最大线程池大小maximumPoolSize）。
 // 和Executors.newCachedThreadPool()一样的
 ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
 new SynchronousQueue<Runnable>());
 testCommon(threadPoolExecutor);
 // 预计结果：
 // 1、 线程池线程数量为：15，超出数量的任务，其他的进入队列中等待被执行
 // 2、 所有任务执行结束，60秒后，如果无任务可执行，所有线程全部被销毁，池的大小恢复为0
 Thread.sleep(60000L);
 System.out.println("60秒后，再看线程池中的数量：" + threadPoolExecutor.getPoolSize());
 }
 /**
 * 5、 定时执行线程池信息：3秒后执行，一次性任务，到点就执行 <br/>
 * 核心线程数量5，最大数量Integer.MAX_VALUE，DelayedWorkQueue延时队列，超出核心线程数量的线程存活时间：0秒
 *
 * @throws Exception
 */ private void threadPoolExecutorTest5() throws Exception {
 // 和Executors.newScheduledThreadPool()一样的
 ScheduledThreadPoolExecutor threadPoolExecutor = new ScheduledThreadPoolExecutor(5);
 threadPoolExecutor.schedule(new Runnable() {
 public void run() {
 System.out.println("任务被执行，现在时间：" + System.currentTimeMillis());
 }
 }, 3000, TimeUnit.MILLISECONDS);
 System.out.println(
 "定时任务，提交成功，时间是：" + System.currentTimeMillis() + ", 当前线程池中线程数量：" + threadPoolExecutor.getPoolSize());
 // 预计结果：任务在3秒后被执行一次
 }
 /**
 * 6、 定时执行线程池信息：线程固定数量5 ，<br/>
 * 核心线程数量5，最大数量Integer.MAX_VALUE，DelayedWorkQueue延时队列，超出核心线程数量的线程存活时间：0秒
 *
 * @throws Exception
 */ private void threadPoolExecutorTest6() throws Exception {
 ScheduledThreadPoolExecutor threadPoolExecutor = new ScheduledThreadPoolExecutor(5);
 // 周期性执行某一个任务，线程池提供了两种调度方式，这里单独演示一下。测试场景一样。
 // 测试场景：提交的任务需要3秒才能执行完毕。看两种不同调度方式的区别
 // 效果1： 提交后，2秒后开始第一次执行，之后每间隔1秒，固定执行一次(如果发现上次执行还未完毕，则等待完毕，完毕后立刻执行)。
 // 也就是说这个代码中是，3秒钟执行一次（计算方式：每次执行三秒，间隔时间1秒，执行结束后马上开始下一次执行，无需等待）
 threadPoolExecutor.scheduleAtFixedRate(new Runnable() {
 public void run() {
 try {
 Thread.sleep(3000L);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("任务-1 被执行，现在时间：" + System.currentTimeMillis());
 }
 }, 2000, 1000, TimeUnit.MILLISECONDS);
 // 效果2：提交后，2秒后开始第一次执行，之后每间隔1秒，固定执行一次(如果发现上次执行还未完毕，则等待完毕，等上一次执行完毕后再开始计时，等待1秒)。
 // 也就是说这个代码钟的效果看到的是：4秒执行一次。 （计算方式：每次执行3秒，间隔时间1秒，执行完以后再等待1秒，所以是 3+1）
 threadPoolExecutor.scheduleWithFixedDelay(new Runnable() {
 public void run() {
 try {
 Thread.sleep(3000L);
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 System.out.println("任务-2 被执行，现在时间：" + System.currentTimeMillis());
 }
 }, 2000, 1000, TimeUnit.MILLISECONDS);
 }
 /**
 * 7、 终止线程：线程池信息： 核心线程数量5，最大数量10，队列大小3，超出核心线程数量的线程存活时间：5秒， 指定拒绝策略的
 *
 * @throws Exception
 */ private void threadPoolExecutorTest7() throws Exception {
 // 创建一个 核心线程数量为5，最大数量为10,等待队列最大是3 的线程池，也就是最大容纳13个任务。
 // 默认的策略是抛出RejectedExecutionException异常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
 ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS,
 new LinkedBlockingQueue<Runnable>(3), new RejectedExecutionHandler() {
 public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
 System.err.println("有任务被拒绝执行了");
 }
 });
 // 测试： 提交15个执行时间需要3秒的任务，看超过大小的2个，对应的处理情况
 for (int i = 0; i < 15; i++) {
 int n = i;
 threadPoolExecutor.submit(new Runnable() {
 public void run() {
 try {
 System.out.println("开始执行：" + n);
 Thread.sleep(3000L);
 System.err.println("执行结束:" + n);
 } catch (InterruptedException e) {
 System.out.println("异常：" + e.getMessage());
 }
 }
 });
 System.out.println("任务提交成功 :" + i);
 }
 // 1秒后终止线程池
 Thread.sleep(1000L);
 threadPoolExecutor.shutdown();
 // 再次提交提示失败
 threadPoolExecutor.submit(new Runnable() {
 public void run() {
 System.out.println("追加一个任务");
 }
 });
 // 结果分析
 // 1、 10个任务被执行，3个任务进入队列等待，2个任务被拒绝执行
 // 2、调用shutdown后，不接收新的任务，等待13任务执行结束
 // 3、 追加的任务在线程池关闭后，无法再提交，会被拒绝执行
 }
 /**
 * 8、 立刻终止线程：线程池信息： 核心线程数量5，最大数量10，队列大小3，超出核心线程数量的线程存活时间：5秒， 指定拒绝策略的
 *
 * @throws Exception
 */ private void threadPoolExecutorTest8() throws Exception {
 // 创建一个 核心线程数量为5，最大数量为10,等待队列最大是3 的线程池，也就是最大容纳13个任务。
 // 默认的策略是抛出RejectedExecutionException异常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
 ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS,
 new LinkedBlockingQueue<Runnable>(3), new RejectedExecutionHandler() {
 public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
 System.err.println("有任务被拒绝执行了");
 }
 });
 // 测试： 提交15个执行时间需要3秒的任务，看超过大小的2个，对应的处理情况
 for (int i = 0; i < 15; i++) {
 int n = i;
 threadPoolExecutor.submit(new Runnable() {
 public void run() {
 try {
 System.out.println("开始执行：" + n);
 Thread.sleep(3000L);
 System.err.println("执行结束:" + n);
 } catch (InterruptedException e) {
 System.out.println("异常：" + e.getMessage());
 }
 }
 });
 System.out.println("任务提交成功 :" + i);
 }
 // 1秒后终止线程池
 Thread.sleep(1000L);
 List<Runnable> shutdownNow = threadPoolExecutor.shutdownNow();
 // 再次提交提示失败
 threadPoolExecutor.submit(new Runnable() {
 public void run() {
 System.out.println("追加一个任务");
 }
 });
 System.out.println("未结束的任务有：" + shutdownNow.size());
 // 结果分析
 // 1、 10个任务被执行，3个任务进入队列等待，2个任务被拒绝执行
 // 2、调用shutdownnow后，队列中的3个线程不再执行，10个线程被终止
 // 3、 追加的任务在线程池关闭后，无法再提交，会被拒绝执行
 }
 public static void main(String[] args) throws Exception {
 new Demo9().threadPoolExecutorTest1();
// new Demo9().threadPoolExecutorTest2();
// new Demo9().threadPoolExecutorTest3();
// new Demo9().threadPoolExecutorTest4();
// new Demo9().threadPoolExecutorTest5();
// new Demo9().threadPoolExecutorTest6();
// new Demo9().threadPoolExecutorTest7();
// new Demo9().threadPoolExecutorTest8();
 }
}

（八）线程数量定义

如何确定合适的数量的线程？

  1. 计算型：CPU数量的1-2倍 （纯内存的运算，加减乘除等任务）
  2. IO型任务：相对比计算型任务，需多开一些线程，要根据具体的IO阻塞线程数量进行考量决定的（例如读取数据库数据较慢的时候，还有就是通过网络接口请求也会消耗很多时间）
  3. tomcat 默认200线程数，

在正式环境当CPU达到80%的时候，就已经达到了充分利用了，如果CPU小于80%，说明利用率是不合理的，上边说的主要是物理主机，虚机50%都要注意了。如果超过此数据值说明需要注意新线程的开启。

（）面试常问技术点

  • ① 线程池的原理:

来看一下线程池究竟是怎么一回事？其实线程池的原理很简单，类似于操作系统中的缓冲区的概念，它的流程如下：先启动若干数量的线程，并让这些线程都处于睡眠状态，当客户端有一个新请求时，就会唤醒线程池中的某一个睡眠线程，让它来处理客户端的这个请求，当处理完这个请求后，线程又处于睡眠状态。可能你也许会问：为什么要搞得这么麻烦，如果每当客户端有新的请求时，我就创建一个新的线程不就完了？这也许是个不错的方法，因为它能使得你编写代码相对容易一些，但你却忽略了一个重要的问题??性能！一个省级数据大集中的银行网络中心，高峰期每秒的客户端请求并发数超过100，如果为每个客户端请求创建一个新线程的话，那耗费的CPU时间和内存将是惊人的，如果采用一个拥有200个线程的线程池，那将会节约大量的的系统资源，使得更多的CPU时间和内存用来处理实际的商业应用，而不是频繁的线程创建与销毁。

  • ② 数据库连接池:

数据库连接是一种关键的有限的昂贵的资源，这一点在多用户的网页应用程序中体现得尤为突出。

一个数据库连接对象均对应一个物理数据库连接，每次操作都打开一个物理连接，使用完都关闭连接，这样造成系统的 性能低下。 数据库连接池的解决方案是在应用程序启动时建立足够的数据库连接，并讲这些连接组成一个连接池(简单说：在一个“池”里放了好多半成品的数据库联接对象)，由应用程序动态地对池中的连接进行申请、使用和释放。对于多于连接池中连接数的并发请求，应该在请求队列中排队等待。并且应用程序可以根据池中连接的使用率，动态增加或减少池中的连接数。

连接池技术尽可能多地重用了消耗内存地资源，大大节省了内存，提高了服务器地服务效率，能够支持更多的客户服务。通过使用连接池，将大大提高程序运行效率，同时，我们可以通过其自身的管理机制来监视数据库连接的数量、使用情况等。

1.最小连接数是连接池一直保持的数据库连接，所以如果应用程序对数据库连接的使用量不大，将会有大量的数据库连接资源被浪费；

2.最大连接数是连接池能申请的最大连接数，如果数据库连接请求超过此数，后面的数据库连接请求将被加入到等待队列中，这会影响之后的数据库操作。

PS：今天就把线程池的原理，运作机制，API代码的演示做了完整的阐述，希望老铁在这里能拿下来代码练一练有所收获。

标签:
打赏 点赞(3)

文章来源：智云一二三科技

文章标题：Java高并发之线程池详解

文章地址：https://www.zhihuclub.com/197617.shtml

关于作者: 智云科技

相关文章

热门文章

1分享新浪图床上传接口源码

点赞(988)阅读(62,256)

2PHP简单实现路由Route功能

点赞(1.20K)阅读(36,001)

3Tideways、xhprof 和 xhgui 打造 PHP 非侵入式监控平台

点赞(425)阅读(4,331)

4centos系统如何查看是否安装了mysql

点赞(593)阅读(3,817)

5curl 工具简述

点赞(582)阅读(3,009)
网站地图