并发编程——线程池

简介: 为了避免频繁创建和销毁线程造成不必要的性能,一般在使用线程时,会采用线程池。

@[TOC]

线程池介绍

Java构建线程的方式

  • new Thread
  • new Runnable
  • new Callable

为了避免频繁创建和销毁线程造成不必要的性能,一般在使用线程时,会采用线程池。

线程池使用方式:

public static void main(String[] args) {
   
    // 线程池的核心线程数如何设置
    // 任务可以分为两种:CPU密集,IO密集。
    ThreadPoolExecutor executor = new ThreadPoolExecutor(
            1,
            2,
            1,
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(1),
            new ThreadFactory() {
   
                @Override
                public Thread newThread(Runnable r) {
   
                    Thread t = new Thread(r);
                    // ...
                    return t;
                }
            },
            new ThreadPoolExecutor.AbortPolicy()
    );

    executor.execute(任务);
    executor.submit(有返回结果的任务);
}

线程池提供的拒绝策略:

  • AbortPolicy:抛出异常。
  • CallerRunsPolicy:让提交任务的线程处理这个任务。
  • DiscardPolicy:丢弃任务
  • DiscardOldestPolicy:扔掉队列最前面的任务,尝试把当前任务添加进去。

    线程池核心属性

// AtomicInteger,就是一个int,写操作用CAS实现,保证了原子性
// ctl维护线程池的2个核心内容:
// 1:线程池状态(高3位,维护着线程池状态)
// 2:工作线程数量(核心线程+非核心线程,低29位,维护着工作线程个数)
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// COUNT_BITS=29
private static final int COUNT_BITS = Integer.SIZE - 3;
// 工作线程的最大个数
// 00100000 00000000 00000000 00000000 - 1
// 000111111111111111111111111111111  
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;


private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

// 拿到线程池状态
private static int runStateOf(int c)     {
    return c & ~CAPACITY; }
// 拿到工作线程个数
private static int workerCountOf(int c)  {
    return c & CAPACITY; }

线程池状态

  1. RUNNING:正常运行的状态,接受新的任务,处理等待队列中的任务。线程池的初始化状态是RUNNING。一旦线程池被创建,就处于RUNNING状态,并且线程池中的任务数为0。
  2. SHUTDOWN:不接受新的任务提交,但是会继续处理等待队列中的任务。调用线程池的shutdown()方法时,线程池由RUNNING变为SHUTDOWN。
  3. STOP:不接受新的任务提交,不再处理等待队列中的任务,中断正在执行任务的线程。调用线程池的shutdownNow()方法时,线程池由(RUNNING or SHUTDOWN)变为STOP。
  4. TIDYING:所有的任务都销毁了,workCount为0,线程池的状态在转换为TIDYING状态时,会执行钩子方法terminated()。
  5. TERMINATED:线程池彻底终止。

核心方法

execute方法

通过execute方法,可以看到线程池处理任务的整体执行流程

public void execute(Runnable command) {
   

    if (command == null)
        throw new NullPointerException();
    // 拿到ctl
    int c = ctl.get();
    // 通过ctl获取当前工作线程个数
    if (workerCountOf(c) < corePoolSize) {
   
        // true:代表是核心线程,false:代表是非核心线程
        if (addWorker(command, true))
            // 如果添加核心线程成功,return结束掉
            return;
        // 如果添加失败,重新获取ctl
        c = ctl.get();
    }
    // 核心线程数已经到了最大值、添加时,线程池状态变为SHUTDOWN/STOP
    // 判断线程池是否是运行状态 && 添加任务到工作队列
    if (isRunning(c) && workQueue.offer(command)) {
   
        // 再次获取ctl的值
        int recheck = ctl.get();
        // 再次判断线程池状态。  DCL
        // 如果状态不是RUNNING,把任务从工作队列移除。
        if (! isRunning(recheck) && remove(command))
            // 走一波拒绝策略。
            reject(command);
        // 线程池状态是RUNNING。
        // 判断工作线程数是否是0个。
        // 可以将核心线程设置为0,所有工作线程都是非核心线程。
        // 核心线程也可以通过keepAlived超时被销毁,所以如果恰巧核心线程被销毁,也会出现当前效果
        else if (workerCountOf(recheck) == 0)
            // 添加空任务的非核心线程去处理工作队列中的任务
            addWorker(null, false);
    }
    // 可能工作队列中的任务存满了,没添加进去,到这就要添加非核心线程去处理任务
    else if (!addWorker(command, false))
        // 执行拒绝策略!
        reject(command);
}

这个方法的主要逻辑是:

  1. 如果当前运行的线程数少于corePoolSize,尝试启动一个新线程并将给定的任务作为它的第一个任务。这个调用会原子性地检查runState和workerCount,以防止在不应该添加线程时发出错误警报。
  2. 如果任务可以成功排队,那么我们仍然需要再次检查是否应该添加一个新线程(因为自上次检查以来,现有的线程可能已经死亡),或者自进入此方法以来,线程池是否已经关闭。所以我们重新检查状态,并在必要时回滚排队,如果已停止,则启动一个新线程。
  3. 如果我们无法排队任务,那么我们尝试添加一个新线程。如果失败,我们知道已经关闭或饱和,所以拒绝任务。

    addWorker添加工作线程

private boolean addWorker(Runnable firstTask, boolean core) {
   
    xxx:
    for (;;) {
   

        int c = ctl.get();
        int rs = runStateOf(c);
        // 判断线程池状态
        if (rs >= SHUTDOWN &&
              // 判断如果线程池的状态为SHUTDOWN,还要处理工作队列中的任务
              // 如果你添加工作线程的方式,是任务的非核心线程,并且工作队列还有任务
            ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty()))
            return false;

        // 判断工作线程个数
        for (;;) {
   

            int wc = workerCountOf(c);
            // 判断1:工作线程是否已经 == 工作线程最大个数
            // 判断2-true判断:判断是核心线程么?如果是判断是否超过核心线程个数
            // 判断2-false判断:如果是非核心线程,查看是否超过设置的最大线程数
            if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            // 对工作线程进行 + 1操作
            if (compareAndIncrementWorkerCount(c))
                // +1成功,跳出外层循环,执行添加工作线程的业务
                // 以CAS方式,对ctl+1,多线程并发操作,只有会有一个成功
                break xxx;
            // 重新拿ctl,
            c = ctl.get();
            // 判断线程池状态是否有变化
            if (runStateOf(c) != rs)
                continue xxx;
        }
    }

    // 添加工作线程的业务  
    boolean workerStarted = false;
    boolean workerAdded = false;
    // Worker就是工作线程
    Worker w = null;
    try {
   
        // 创建工作线程,将任务传到Worker中
        w = new Worker(firstTask);
        final Thread t = w.thread;
        // 只有你写的线程工厂返回的是null,这里才会为null
        if (t != null) {
   
            // 获取锁资源
            final ReentrantLock mainLock = this.mainLock;
            // 加锁。  因为我要在启动这个工作线程时,避免线程池状态发生变化,加锁。
            mainLock.lock();
            try {
   
                // 重新获取ctl,拿到线程池状态
                int rs = runStateOf(ctl.get());
                // DCL i think you know~~~
                if (rs < SHUTDOWN ||
                    (rs == SHUTDOWN && firstTask == null)) {
   
                       // 判断Worker中的thread是否已经启动了,一般不会启动,除非你在线程工厂把他启动了
                    if (t.isAlive()) 
                        throw new IllegalThreadStateException();
                    // 将工作线程存储到hashSet中
                    workers.add(w);
                    // 获取工作线程个数,判断是否需要修改最大工作线程数记录。
                    int s = workers.size();
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                    // 工作线程添加成功     0
                    workerAdded = true;
                }
            } finally {
   
                mainLock.unlock();
            }
            // 如果添加成功
            if (workerAdded) {
   
                // 启动工作线程
                t.start();
                // 设置标识为true
                workerStarted = true;
            }
        }
    } finally {
   
        // 如果工作线程启动失败
        if (! workerStarted)
            addWorkerFailed(w);
    }
    return workerStarted;
}

// 如果添加工作线程失败,执行
private void addWorkerFailed(Worker w) {
   
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
   
        // 说明worker可能存放到了workers的hashSet中。
        if (w != null)
            // 移除!
            workers.remove(w);
        // 减掉workerCount的数值 -1
        decrementWorkerCount();
        // 尝试干掉自己
        tryTerminate();
    } finally {
   
        mainLock.unlock();
    }
}

这个方法的主要逻辑是:

  1. 检查是否正在关闭或已经关闭,如果是,根据是否有任务以及队列是否为空来决定是否返回false。
  2. 如果没有关闭,那么检查当前线程数是否超过了核心线程数或最大线程数,如果超过了则返回false。
  3. 如果没有超过线程数限制,那么通过CAS操作增加workerCount,如果成功则跳出循环。如果失败,则重新读取ctl并重新计算runState,如果runState发生变化则继续循环。如果未发生变化,则继续尝试增加workerCount。
  4. 创建Worker对象,并启动对应的线程。如果启动成功,则返回true。如果启动失败,则调用addWorkerFailed()方法处理失败情况。

    runWorker执行任务

final void runWorker(Worker w) {
   
    // 拿到当前线程对象
    Thread wt = Thread.currentThread();
    // 拿到worker中存放的Runnable
    Runnable task = w.firstTask;
    // 将worker中的任务清空
    w.firstTask = null;
    boolean completedAbruptly = true;
    try {
   
        // 如果Worker自身携带任务,直接执行
        // 如果Worker携带的是null,通过getTask去工作队列获取任务
        while (task != null || (task = getTask()) != null) {
   
            w.lock();
            // 判断线程池状态是否大于等于STOP,如果是要中断当前线程
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 // 中断当前线程(DCL)
                 (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);
    }
}

这个方法的主要逻辑是:

  1. 获取当前线程和任务。如果任务不为空,则执行任务。如果任务为空,则从任务队列中获取任务。如果获取任务失败,则退出循环。
  2. 在执行任务前,检查线程是否中断,如果中断则停止执行任务。如果未中断,则执行任务。如果抛出运行时异常或错误,则捕获并记录异常。如果抛出其他异常,则转换为错误并抛出。
  3. 在执行任务后,将任务设置为null,增加已完成任务数,并解锁worker。
  4. 如果循环结束时,任务为空,则说明任务队列已关闭或已空,将completedAbruptly设置为false。否则,说明任务队列中还有任务,将completedAbruptly保持为true。
  5. 最后,处理worker退出,根据completedAbruptly的值决定是否重置线程池状态。

    getTask工作线程排队拿任务

private Runnable getTask() {
   
    boolean timedOut = false; // Did the last poll() time out?

    for (;;) {
   
        int c = ctl.get();
        int rs = runStateOf(c);

        // 线程池状态判断
        // 如果线程池状态为SHUTDOWN && 工作队列为空
        // 如果线程池状态为STOP
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
   
            // 对工作线程个数--
            decrementWorkerCount();
            return null;
        }

        // 对数量的判断。
        int wc = workerCountOf(c);

        // 判断核心线程是否允许超时?
        // 工作线程个数是否大于核心线程数
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;

        // 判断工作线程是否超过了最大线程数 && 工作队列为null
        if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) {
   
            // 工作线程数有问题,必须-1,干掉当前工作线程
            // 工作线程是否超过了核心线程,如果超时,就干掉当前线程
            // 对工作线程个数--
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }

        try {
   
            // 如果是非核心,走poll,拉取工作队列任务,
            // 如果是核心线程,走take一直阻塞,拉取工作队列任务
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                // 当工作队列没有任务时,这时就会被Condition通过await阻塞线程
                // 当有任务添加到工作线程后,这是添加完任务后,就会用过Condition.signal唤醒阻塞的线程
                workQueue.take();
            if (r != null)
                return r;
            // 执行的poll方法,并且在指定时间没拿到任务,
            timedOut = true;
        } catch (InterruptedException retry) {
   
            timedOut = false;
        }
    }
}

该方法的主要逻辑是:

  1. 检查线程池状态,如果线程池已经关闭或停止,并且任务队列为空,则减少worker数量并返回null。
  2. 检查worker数量是否超过了最大线程数或核心线程数,如果是,则减少worker数量并返回null。
  3. 如果worker数量未超过最大线程数或核心线程数,则从任务队列中获取任务。如果获取任务失败,则返回null。如果获取任务成功,则返回任务。
  4. 如果在获取任务时发生中断,则重新尝试获取任务。如果在指定的时间内无法获取任务,则将timedOut标志设置为true。如果在指定的时间内无法获取任务并且允许超时,则减少worker数量并返回null。
相关文章
|
27天前
|
Java 程序员 调度
【JAVA 并发秘籍】进程、线程、协程:揭秘并发编程的终极武器!
【8月更文挑战第25天】本文以问答形式深入探讨了并发编程中的核心概念——进程、线程与协程,并详细介绍了它们在Java中的应用。文章不仅解释了每个概念的基本原理及其差异,还提供了实用的示例代码,帮助读者理解如何在Java环境中实现这些并发机制。无论你是希望提高编程技能的专业开发者,还是准备技术面试的求职者,都能从本文获得有价值的见解。
33 1
|
1月前
|
Java 开发者
解锁并发编程新姿势!深度揭秘AQS独占锁&ReentrantLock重入锁奥秘,Condition条件变量让你玩转线程协作,秒变并发大神!
【8月更文挑战第4天】AQS是Java并发编程的核心框架,为锁和同步器提供基础结构。ReentrantLock基于AQS实现可重入互斥锁,比`synchronized`更灵活,支持可中断锁获取及超时控制。通过维护计数器实现锁的重入性。Condition接口允许ReentrantLock创建多个条件变量,支持细粒度线程协作,超越了传统`wait`/`notify`机制,助力开发者构建高效可靠的并发应用。
72 0
|
11天前
|
缓存 监控 Java
Java中的并发编程:理解并应用线程池
在Java的并发编程中,线程池是提高应用程序性能的关键工具。本文将深入探讨如何有效利用线程池来管理资源、提升效率和简化代码结构。我们将从基础概念出发,逐步介绍线程池的配置、使用场景以及最佳实践,帮助开发者更好地掌握并发编程的核心技巧。
|
1月前
|
数据采集 Java Python
Python并发编程:多线程(threading模块)
Python是一门强大的编程语言,提供了多种并发编程方式,其中多线程是非常重要的一种。本文将详细介绍Python的threading模块,包括其基本用法、线程同步、线程池等,最后附上一个综合详细的例子并输出运行结果。
|
30天前
|
数据采集 Java Python
Python并发编程:多线程(threading模块)
本文详细介绍了Python的threading模块,包括线程的创建、线程同步、线程池的使用,并通过多个示例展示了如何在实际项目中应用这些技术。通过学习这些内容,您应该能够熟练掌握Python中的多线程编程,提高编写并发程序的能力。 多线程编程可以显著提高程序的并发性能,但也带来了新的挑战和问题。在使用多线程时,需要注意避免死锁、限制共享资源的访问,并尽量使用线程池来管理和控制线程。
|
1月前
|
缓存 Java 数据处理
Java中的并发编程:解锁多线程的力量
在Java的世界里,并发编程是提升应用性能和响应能力的关键。本文将深入探讨Java的多线程机制,从基础概念到高级特性,逐步揭示如何有效利用并发来处理复杂任务。我们将一起探索线程的创建、同步、通信以及Java并发库中的工具类,带你领略并发编程的魅力。
|
21天前
|
Java 数据库连接 微服务
揭秘微服务架构下的数据魔方:Hibernate如何玩转分布式持久化,实现秒级响应的秘密武器?
【8月更文挑战第31天】微服务架构通过将系统拆分成独立服务,提升了可维护性和扩展性,但也带来了数据一致性和事务管理等挑战。Hibernate 作为强大的 ORM 工具,在微服务中发挥关键作用,通过二级缓存和分布式事务支持,简化了对象关系映射,并提供了有效的持久化策略。其二级缓存机制减少数据库访问,提升性能;支持 JTA 保证跨服务事务一致性;乐观锁机制解决并发数据冲突。合理配置 Hibernate 可助力构建高效稳定的分布式系统。
38 0
|
21天前
|
程序员 调度 C++
解锁Ruby并发编程新境界!Fiber与线程:轻量级VS重量级,你选哪一派引领未来?
【8月更文挑战第31天】Ruby提供了多种并发编程方案,其中Fiber与线程是关键机制。Fiber是自1.9版起引入的轻量级并发模型,无需独立堆栈和上下文切换,由程序员控制调度。线程则为操作系统级别,具备独立堆栈和上下文,能利用多核处理器并行执行。通过示例代码展示了Fiber和线程的应用场景,如任务调度和多URL数据下载,帮助开发者根据需求选择合适的并发模型,提升程序性能与响应速度。
26 0
|
1月前
|
Java API 开发者
Java中的并发编程:解锁多线程的潜力
在数字化时代的浪潮中,并发编程已成为软件开发的核心技能之一。本文将深入探讨Java中的并发编程概念,通过实例分析与原理解释,揭示如何利用多线程提升应用性能和响应性。我们将从基础的线程创建开始,逐步过渡到高级的同步机制,并探讨如何避免常见的并发陷阱。读者将获得构建高效、稳定并发应用所需的知识,同时激发对Java并发更深层次探索的兴趣。
33 2
|
29天前
|
Java C语言 C++
并发编程进阶:线程同步与互斥
并发编程进阶:线程同步与互斥
29 0