当前位置： 首页 > news >正文

【＜Java中的线程池＞】

news 来源：原创 2024/4/24 13:57:23

Java中的线程池

          • [1] 使用线程池的好处
          • [2] Executor框架
          • [3] ThreadPoolExecutor类的参数字段
          • [4] 线程池的排队策略
          • [6] 拒绝策略
          • [5] 常见的阻塞队列
          • [6] Java提供的四种线程池
          • [7] 手写一个线程池

[1] 使用线程池的好处

为了降低资源消耗，提高响应速度，提高线程的可管理性，所以出现了线程池。

  1. 降低资源消耗：通过重复利用已创建的线程降低线程创建和销毁造成的消耗;
  2. 提高响应速度：当任务到达时，任务可以不需要的等到线程创建就能立即执行;
  3. 提高线程的可管理性：线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。
[2] Executor框架

https://www.cnblogs.com/zxrxzw/p/12116209.html)
1. 什么是Executor框架？

我们知道线程池就是线程的集合，线程池集中管理线程，以实现线程的重用，降低资源消耗，提高响应速度等。线程用于执行异步任务，单个的线程既是工作单元也是执行机制。从JDK1.5开始，为了把工作单元与执行机制分离开，Executor框架是一个统一创建与运行的接口。Executor框架实现的就是线程池的功能。

2. Executor框架结构图解

Executor框架包括3大部分：

（1）任务：也就是工作单元，包括被执行任务需要实现的接口：Runnable接口或者Callable接口；

（2）任务的执行：也就是把任务分派给多个线程的执行机制，包括Executor接口及继承自Executor接口的ExecutorService接口。

（3）异步计算的结果：包括Future接口及实现了Future接口的FutureTask类。

3. Executor框架成员：

ThreadPoolExecutor实现类、ScheduledThreadPoolExecutor实现类、Future接口、Runnable和Callable接口、Executors工厂类

img
[3] ThreadPoolExecutor类的参数字段
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)
  • corePoolSize：核心线程数：线程池中会维护一个最小的线程数量，即使这些线程处于空闲状态，他们也不会被销毁（除非设置了allowCoreThreadTimeOut），而且线程数小于此值的话，仍然会创建新的线程。
  • maximumPoolSize：最大线程数：线程池所能容的最多线程数量。一个任务被提交到线程池后，首先会缓存到工作队列中，如果工作队列满了，则会创建一个新线程，然后从工作队列中的取出一个任务交由新线程来处理，而将刚提交的任务放入工作队列。线程池不会无限制的去创建新线程，它会有一个最大线程数量的限制，这个数量即由maximunPoolSize来指定。
  • keepAliveTime ：线程空闲的保活时间：一个线程如果处于空闲状态，并且当前的线程数量大于corePoolSize，那么在指定时间后，这个空闲线程会被销毁，这里的指定时间由keepAliveTime来设定。
  • unit：时间单位：keepAliveTime的计量单位。
  • workQueue：存储线程的队列：新任务被提交后，会先进入到此工作队列中，任务调度时再从队列中取出任务。jdk中提供了四种工作队列。
  • threadFactory：创建线程的工厂：创建一个新线程时使用的工厂，可以用来设定线程名、是否为daemon线程等等
  • handler：拒绝策略：当工作队列中的任务已到达最大限制，并且线程池中的线程数量也达到最大限制，这时如果有新任务提交进来，该如何处理呢。这里的拒绝策略，就是解决这个问题的，jdk中提供了4中拒绝策略。
[4] 线程池的排队策略
  1. 当线程数达到 小于corePoolSize ，即使有空闲队列，线程池也会优先创建一个新线程来执行任务。
  2. 当线程数达到 corePoolSize 且并没有线程空闲时，加入任务，这时候会有两种情况：
    • 如果队列选择了无界队列，新加的任务会被加入workQueue 队列排队，直到有空闲的线程。
    • 如果队列选择了有界队列，任务会优先入队，那么任务超过了队列大小时，会创建 maximumPoolSize - corePoolSize 数目的线程来救急，若此范围线程空闲，则有keepAliveTime决定存活时间。
  3. 针对有界队列，如果线程到达 maximumPoolSize 仍然有新任务这时会执行拒绝策略。拒绝策略 jdk 提供了 4 种实现，其它著名框架也提供了实现。
  4. 针对有界队列，当高峰过去后，超过corePoolSize 的救急线程如果一段时间没有任务做，需要结束节省资源，这个时间由keepAliveTime 和 unit 来控制。
[6] 拒绝策略

四种线程池拒绝策略

当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize时，如果还有任务到来就会采取任务拒绝策略，通常有以下四种策略：

  • ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
  • ThreadPoolExecutor.DiscardPolicy：丢弃任务，但是不抛出异常。
  • ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新提交被拒绝的任务
  • ThreadPoolExecutor.CallerRunsPolicy：由调用线程（提交任务的线程）处理该任务

框架实现：

  • Dubbo 的实现，在抛出 RejectedExecutionException 异常之前会记录日志，并 dump 线程栈信息，方
    便定位问题
  • Netty 的实现，是创建一个新线程来执行任务
  • ActiveMQ 的实现，带超时等待（60s）尝试放入队列，类似我们之前自定义的拒绝策略
  • PinPoint 的实现，它使用了一个拒绝策略链，会逐一尝试策略链中每种拒绝策略
[5] 常见的阻塞队列

https://blog.csdn.net/xiewenfeng520/article/details/107142566

前面我们介绍了线程池内部有一个排队策略，任务可能需要在队列中进行排队等候。常见的阻塞队列包括如下的四种，接下来我们一起来看看吧。

  • ArrayBlockingQueue：基于数组的有界阻塞队列，按FIFO排序。新任务进来后，会放到该队列的队尾，有界的数组可以防止资源耗尽问题。当线程池中线程数量达到corePoolSize后，再有新任务进来，则会将任务放入该队列的队尾，等待被调度。如果队列已经是满的，则创建一个新线程，如果线程数量已经达到maxPoolSize，则会执行拒绝策略。

  • LinkedBlockingQuene：基于链表的无界阻塞队列（其实最大容量为Interger.MAX），按照FIFO排序。由于该队列的近似无界性，当线程池中线程数量达到corePoolSize后，再有新任务进来，会一直存入该队列，而不会去创建新线程直到maxPoolSize，因此使用该工作队列时，参数maxPoolSize其实是不起作用的。

  • SynchronousQuene：一个不缓存任务的阻塞队列，生产者放入一个任务必须等到消费者取出这个任务。也就是说新任务进来时，不会缓存，而是直接被调度执行该任务，如果没有可用线程，则创建新线程，如果线程数量达到maxPoolSize，则执行拒绝策略。

  • PriorityBlockingQueue：具有优先级的无界阻塞队列，优先级通过参数Comparator实现。

阻塞队列特点及其使用场景：

ArrayBlockingQueue（有界）:

  • 内部使用一个数组作为其存储空间，数组的存储空间是预先分配的
  • 优点是 put 和 take操作不会增加GC的负担（因为空间是预先分配的）
  • 缺点是 put 和 take操作使用同一个锁，可能导致锁争用，导致较多的上下文切换。
  • 会执行拒绝策略，有救急线程
  • ArrayBlockingQueue适合在生产者线程和消费者线程之间的并发程度较低的情况下使用。

LinkedBlockingQueue：

  • 是一个无界队列（其实队列长度是Integer.MAX_VALUE）
  • 内部存储空间是一个链表，并且链表节点所需的存储空间是动态分配的
  • 优点是 put 和 take 操作使用两个显式锁（putLock和takeLock）
  • 缺点是增加了GC的负担，因为空间是动态分配的。
  • 不会执行拒绝策略，无救急线程
  • LinkedBlockingQueue适合在生产者线程和消费者线程之间的并发程度较高的情况下使用。

SynchronousQueue：

  • SynchronousQueue可以被看做一种特殊的有界队列。
  • 生产者线程生产一个产品之后，会等待消费者线程来取走这个产品，才会接着生产下一个产品。
  • 适合在生产者线程和消费者线程之间的处理能力相差不大的情况下使用。
[6] Java提供的四种线程池

Executors只是对线程池一些特定情况的简洁使用，直接用ThreadPoolExecutor构造线程池会获得更为强大的功能。java自带的四种线程池通过 ThreadPoolExecutor 的构造方法实现：

  1. newCachedThreadPool：是一个根据需要来创建线程的线程池，它的corePoolSize为0，maximumPoolSize设置为Integer.MAX_VALUE，所以maximumPool是无界的，keepAliveTime设置为60L，代表所有空闲线程等待新任务的最长时间为60L。

这意味着，如果主线程提交任务的速度高于maxiunmPool中线程处理任务的速度时，将会不断CachedThreadPool创建新线程。极端情况下会耗尽CPU资源。所以它适用于负载较轻的场景，执行短期异步任务如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。（可以使得任务快速得到执行，因为任务时间执行短，可以很快结束，也不会造成cpu过度切换）

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
  1. newFixedThreadPool：创建一个可重用固定线程数的线程池，可控制线程最大并发数，超出的线程会在队列中等待。因为采用无界的阻塞队列LinkedBlockingQueue，所以maximumPoolSize和keepAliveTime为无效函数，实际线程数量永远不会变化，只会小于等于corePoolSize。

适用于负载较重的场景，对当前线程数量进行限制。（保证线程数可控，不会造成线程过多，导致系统负载更为严重）

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
}
  1. newSingleThreadExecutor：创建一个单线程的线程池，适用于需要保证顺序执行各个任务。它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。它的corePoolSize和keepAliveTime设置为1，由于采用无界堵塞队列LinkedBlockingQueue，可以保证无限缓存任务，并且线程固定。
public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
}
  1. newScheduledThreadPool：它继承于ThreadPoolExecutor。创建一个定长任务线程池，适用于执行延时或者周期性任务。
public class OneMoreStudy {
    public static void main(String[] args) {
        final SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(3);
        System.out.println("提交时间: " + sdf.format(new Date()));
        scheduledThreadPool.schedule(new Runnable() {
                @Override
                public void run() {
                    System.out.println("运行时间: " + sdf.format(new Date()));
                }
            }, 3, TimeUnit.SECONDS);
        scheduledThreadPool.shutdown();
    }
}
[7] 手写一个线程池

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-8aejg5JF-1596598779485)(X:\Users\xu\AppData\Roaming\Typora\typora-user-images\image-20200709145932351.png)]

package cn.itcast.n8;

import lombok.extern.slf4j.Slf4j;
import org.springframework.core.log.LogDelegateFactory;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j(topic = "c.TestPool")
public class TestPool {
    public static void main(String[] args) {
        ThreadPool threadPool = new ThreadPool(1,
                1000, TimeUnit.MILLISECONDS, 1, (queue, task)->{
            // 1. 死等
//            queue.put(task);
            // 2) 带超时等待
//            queue.offer(task, 1500, TimeUnit.MILLISECONDS);
            // 3) 让调用者放弃任务执行
//            log.debug("放弃{}", task);
            // 4) 让调用者抛出异常
//            throw new RuntimeException("任务执行失败 " + task);
            // 5) 让调用者自己执行任务
            task.run();
        });
        for (int i = 0; i < 4; i++) {
            int j = i;
            threadPool.execute(() -> {
                try {
                    Thread.sleep(1000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.debug("{}", j);
            });
        }
    }
}

@FunctionalInterface // 拒绝策略
interface RejectPolicy<T> {
    void reject(BlockingQueue<T> queue, T task);
}

@Slf4j(topic = "c.ThreadPool")
class ThreadPool {
    // 任务队列
    private BlockingQueue<Runnable> taskQueue;

    // 线程集合
    private HashSet<Worker> workers = new HashSet<>();

    // 核心线程数
    private int coreSize;

    // 获取任务时的超时时间
    private long timeout;

    private TimeUnit timeUnit;

    private RejectPolicy<Runnable> rejectPolicy;

    // 执行任务
    public void execute(Runnable task) {
        // 当任务数没有超过 coreSize 时，直接交给 worker 对象执行
        // 如果任务数超过 coreSize 时，加入任务队列暂存
        synchronized (workers) {
            if(workers.size() < coreSize) {
                Worker worker = new Worker(task);
                log.debug("新增 worker{}, {}", worker, task);
                workers.add(worker);
                worker.start();
            } else {
//                taskQueue.put(task);
                // 1) 死等
                // 2) 带超时等待
                // 3) 让调用者放弃任务执行
                // 4) 让调用者抛出异常
                // 5) 让调用者自己执行任务
                taskQueue.tryPut(rejectPolicy, task);
            }
        }
    }

    public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapcity, RejectPolicy<Runnable> rejectPolicy) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new BlockingQueue<>(queueCapcity);
        this.rejectPolicy = rejectPolicy;
    }

    class Worker extends Thread{
        private Runnable task;

        public Worker(Runnable task) {
            this.task = task;
        }

        @Override
        public void run() {
            // 执行任务
            // 1) 当 task 不为空，执行任务
            // 2) 当 task 执行完毕，再接着从任务队列获取任务并执行
//            while(task != null || (task = taskQueue.take()) != null) {
            while(task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
                try {
                    log.debug("正在执行...{}", task);
                    task.run();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    task = null;
                }
            }
            synchronized (workers) {
                log.debug("worker 被移除{}", this);
                workers.remove(this);
            }
        }
    }
}
@Slf4j(topic = "c.BlockingQueue")
class BlockingQueue<T> {
    // 1. 任务队列
    private Deque<T> queue = new ArrayDeque<>();

    // 2. 锁
    private ReentrantLock lock = new ReentrantLock();

    // 3. 生产者条件变量
    private Condition fullWaitSet = lock.newCondition();

    // 4. 消费者条件变量
    private Condition emptyWaitSet = lock.newCondition();

    // 5. 容量
    private int capcity;

    public BlockingQueue(int capcity) {
        this.capcity = capcity;
    }

    // 带超时阻塞获取
    public T poll(long timeout, TimeUnit unit) {
        lock.lock();
        try {
            // 将 timeout 统一转换为 纳秒
            long nanos = unit.toNanos(timeout);
            while (queue.isEmpty()) {
                try {
                    // 返回值是剩余时间
                    if (nanos <= 0) {
                        return null;
                    }
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        } finally {
            lock.unlock();
        }
    }

    // 阻塞获取
    public T take() {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        } finally {
            lock.unlock();
        }
    }

    // 阻塞添加
    public void put(T task) {
        lock.lock();
        try {
            while (queue.size() == capcity) {
                try {
                    log.debug("等待加入任务队列 {} ...", task);
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("加入任务队列 {}", task);
            queue.addLast(task);
            emptyWaitSet.signal();
        } finally {
            lock.unlock();
        }
    }

    // 带超时时间阻塞添加
    public boolean offer(T task, long timeout, TimeUnit timeUnit) {
        lock.lock();
        try {
            long nanos = timeUnit.toNanos(timeout);
            while (queue.size() == capcity) {
                try {
                    if(nanos <= 0) {
                        return false;
                    }
                    log.debug("等待加入任务队列 {} ...", task);
                    nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("加入任务队列 {}", task);
            queue.addLast(task);
            emptyWaitSet.signal();
            return true;
        } finally {
            lock.unlock();
        }
    }

    public int size() {
        lock.lock();
        try {
            return queue.size();
        } finally {
            lock.unlock();
        }
    }

    public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
        lock.lock();
        try {
            // 判断队列是否满
            if(queue.size() == capcity) {
                rejectPolicy.reject(this, task);
            } else {  // 有空闲
                log.debug("加入任务队列 {}", task);
                queue.addLast(task);
                emptyWaitSet.signal();
            }
        } finally {
            lock.unlock();
        }
    }
}

相关文章：

  • 关于数组添加值和访问值的一些小问题
  • hdu 5623 KK's Number（dp）
  • 【阅读源码系列】HashMap源码分析（JDK1.7和JDK1.8）
  • 使用开源的PullToRefreshScrollView scrollTo和scrollby遇到的问题
  • 【阅读源码系列】ThreadPoolExecutor源码
  • ​猴子吃桃问题：每天都吃了前一天剩下的一半多一个。
  • YII 路由配置
  • 讲一下ISO七层模型？
  • Python 17.1 HTTP协议简介
  • 培训机构出来的iOS学员怎么了？
  • 输入url（网址）之后到显示网页的过程？
  • Java中的陷阱题
  • Host文件?
  • js阻止表单重复提交
  • 为什么要域名解析？
  • FineReport中如何实现自动滚屏效果
  • JavaScript 无符号位移运算符 三个大于号 的使用方法
  • JavaScript类型识别
  • Java新版本的开发已正式进入轨道，版本号18.3
  • JS实现简单的MVC模式开发小游戏
  • LintCode 31. partitionArray 数组划分
  • linux安装openssl、swoole等扩展的具体步骤
  • node 版本过低
  • Object.assign方法不能实现深复制
  • vuex 学习笔记 01
  • 关键词挖掘技术哪家强（一）基于node.js技术开发一个关键字查询工具
  • 关于使用markdown的方法（引自CSDN教程）
  • 后端_ThinkPHP5
  • 每天一个设计模式之命令模式
  • 前端学习笔记之观察者模式
  • 深度学习中的信息论知识详解
  • 深入体验bash on windows，在windows上搭建原生的linux开发环境，酷！
  • 算法-图和图算法
  • Oracle Portal 11g Diagnostics using Remote Diagnostic Agent (RDA) [ID 1059805.
  • 智能情侣枕Pillow Talk，倾听彼此的心跳
  • 昨天1024程序员节，我故意写了个死循环~
  • ​MySQL主从复制一致性检测
  • #ifdef 的技巧用法
  • $分析了六十多年间100万字的政府工作报告，我看到了这样的变迁
  • （11）MATLAB PCA+SVM 人脸识别
  • （翻译）Entity Framework技巧系列之七 - Tip 26 – 28
  • （附源码）ssm教材管理系统 毕业设计 011229
  • （附源码）ssm考生评分系统 毕业设计 071114
  • （附源码）ssm学生管理系统 毕业设计 141543
  • （十五）Flask覆写wsgi_app函数实现自定义中间件
  • (转)Linux NTP配置详解 (Network Time Protocol)
  • (转)菜鸟学数据库（三）——存储过程
  • **PyTorch月学习计划 - 第一周；第6-7天: 自动梯度（Autograd）**
  • .NET Core SkiaSharp 替代 System.Drawing.Common 的一些用法
  • .NET Core/Framework 创建委托以大幅度提高反射调用的性能
  • .net 反编译_.net反编译的相关问题
  • .NET 中各种混淆（Obfuscation）的含义、原理、实际效果和不同级别的差异（使用 SmartAssembly）
  • .net 逐行读取大文本文件_如何使用 Java 灵活读取 Excel 内容 ？
  • .NET建议使用的大小写命名原则
  • /proc/stat文件详解（翻译）