热门标签 | HotTags
当前位置:  开发笔记 > 编程语言 > 正文

java线程池的实现原理源码分析

这篇文章主要介绍“java线程池的实现原理源码分析”,在日常操作中,相信很多人在java线程池的实现原理源码分析问题上存在疑惑,小编查阅了各式资

这篇文章主要介绍“java线程池的实现原理源码分析”,在日常操作中,相信很多人在java线程池的实现原理源码分析问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”java线程池的实现原理源码分析”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!

线程池的起源

背景: 随着计算机硬件的升级换代,使我们的软件具备多线程执行任务的能力。当我们在进行多线程编程时,就需要创建线程,如果说程序并发很高的话,我们会创建大量的线程,而每个线程执行一个时间很短的任务就结束了,这样频繁创建线程,会极大的降低系统性能,增加服务器开销,因为创建线程和销毁线程都需要额外的消耗。

这时我们就可以借助池化技术,来优化这一缺陷,线程池就诞生了。

池化技术的本质是在高并发场景下,为了实现资源复用,减少资源创建销毁等开销,如果并发数很小没有明显优势(资源一直占用系统内存,没有机会被使用)。

池化技术介绍: 什么时池化技术呢?池化技术是一种编程技巧,当程序出现高并发时,能够明显的优化程序,降低系统频繁创建销毁连接等额外开销。我们经常接触到的池化技术有数据库连接池、线程池、对象池等等。池化技术的特点是将一些高成本的资源维护在一个特定的池子(内存)中,规定其最小连接数、最大连接数、阻塞队列,溢出规则等配置,方便统一管理。一般情况下也会附带一些监控,强制回收等配套功能。

池化技术作为一种资源使用技术,典型的使用情形是:

  • 获取资源的成本较高的时候

  • 请求资源的频率很高且使用资源总数较低的时候

  • 面对性能问题,涉及到处理时间延迟的时候

池化技术资源分类:

  • 系统调用的系统资源,如线程、进程、内存分配等

  • 网络通信的远程资源, 如数据库连接、套接字连接等

线程池的定义和使用

线程池是我们为了规避创建线程,销毁线程额外开销而诞生的,所以说我们定义创建好线程池之后,就不需要自己来创建线程,而是使用线程池调用执行我们的任务。下面我们一起看一下如何定义并创建线程池。

方案一:Executors(仅做了解,推荐使用方案二)

创建线程池可以使用Executors,其中提供了一系列工厂方法用于创建线程池,返回的线程池都实现了ExecutorService接口。

ExecutorService 接口是Executor接口的子类接口,使用更为广泛,其提供了线程池生命周期管理的方法,返回 Future 对象

也就是说我们通过Executors创建线程池,得到ExecutorService,通过ExecutorService执行异步任务(实现Runnable接口)

Executors 可以创建一下几种类型的线程池:

  • newCachedThreadPool 创建一个可缓存线程池,如果线程池线程数量过剩,会在60秒后回收掉多余线程资源,当任务书增加,线程不够用,则会新建线程。

  • newFixedThreadPool 创建一个定长线程池,可控制线程最大并发数,超出的线程会在队列中等待。

  • newScheduledThreadPool 创建一个定长线程池,支持定时及周期性任务执行。

  • newSingleThreadExecutor 创建一个单线程的线程池,只使用唯一的线程来执行任务,可以保证任务按照提交顺序来完成。

方案二:ThreadPoolExecutor

在阿里巴巴开发规范中,规定线程池不允许通过Executors创建,而是通过ThreadPoolExecutor创建。

好处:让写的同学可以更加明确线程池的运行规则,规避资源耗尽的风险。

ThreadPoolExecutor的七大参数:

(1)corePoolSize 核心线程数量,核心线程会一直保留,不会被销毁。

(2)maximumPoolSize 最大线程数,当核心线程不能满足任务需要时,系统就会创建新的线程来执行任务。

(3)keepAliveTime 存活时间,核心线程之外的线程空闲多长时间就会被销毁。

(4)timeUnit 代表线程存活的时间单位。

(5)BlockingQueue 阻塞队列

  • 如果正在执行的任务超过了最大线程数,可以存放在队列中,当线程池中有空闲资源就可以从队列中取出任务继续执行。

  • 队列类型有如下几种类型:LinkedBlockingQueue ArrayBlockingQueue SynchronousQueue TransferQueue。

(6)threadFactory 线程工厂,用来创建线程的,可以自定义线程,比如我们可以定义线程组名称,在jstack问题排查时,非常有帮助。

(7)rejectedExecutionHandler 拒绝策略,

当所有线程(最大线程数)都在忙,并且任务队列处于满任务的状态,则会执行拒绝策略。

JDK为我们提供了四种拒绝策略,我们必须都得熟悉

  • AbortPolicy: 丢弃任务,并抛出异常RejectedExecutionException。 默认

  • DiscardPolicy: 丢弃最新的任务,不抛异常。

  • DiscardOldestPolicy: 扔掉排队时间最久的任务,也就是最旧的任务。

  • CallerRuns: 由调用者(提交异步任务的线程)处理任务。

线程池的实现原理

想要实现一个线程池我们就需要关心ThreadPoolExecutor类,因为Executors创建线程池也是通过new ThreadPoolExecutor对象。

看一下ThreadPoolExecutor的类继承关系,可以看出为什么通过Executors创建的线程池返回结果是ExecutorService,因为ThreadPoolExecutor是ExecutorService接口的实现类,而Executors创建线程池本质也是创建的ThreadPoolExecutor 对象。

java线程池的实现原理源码分析

下面我们一起看一下ThreadPoolExecutor的源码,首先是ThreadPoolExecutor内定义的变量,常量:

    // 复合类型变量 是一个原子整数  控制状态(运行状态|线程池活跃线程数量)
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); 
    private static final int COUNT_BITS = Integer.SIZE - 3; // 低29位 
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1; // 容量
    // 运行状态存储在高位3位
    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;  // 所有的任务已结束,活跃线程为0,线程过渡到TIDYING状       态,将会执行terminated()钩子方法
    private static final int TERMINATED =  3 << COUNT_BITS;  // terminated()方法已经完成
    // 设置 ctl 参数方法
    private static int runStateOf(int c)     { return c & ~CAPACITY; }
    private static int workerCountOf(int c)  { return c & CAPACITY; }
    private static int ctlOf(int rs, int wc) { return rs | wc; }
    /**
     * 阻塞队列
     */
    private final BlockingQueue workQueue;
    /**
     * Lock 锁.
     */
    private final ReentrantLock mainLock = new ReentrantLock();
    /**
     * 工人们
     */
    private final HashSet workers = new HashSet();
    /**
     * 等待条件支持等待终止
     */
    private final Condition termination = mainLock.newCondition();
    /**
     * 最大的池大小.
     */
    private int largestPoolSize;
    /**
     * 完成任务数
     */
    private long completedTaskCount;
    /**
     * 线程工厂
     */
    private volatile ThreadFactory threadFactory;
    /**
     * 拒绝策略
     */
    private volatile RejectedExecutionHandler handler;
    /**
     * 存活时间
     */
    private volatile long keepAliveTime;
    /**
     * 允许核心线程数
     */
    private volatile boolean allowCoreThreadTimeOut;
    /**
     * 核心线程数
     */
    private volatile int corePoolSize;
    /**
     * 最大线程数
     */
    private volatile int maximumPoolSize;
    /**
     * 默认拒绝策略
     */
    private static final RejectedExecutionHandler defaultHandler =
        new AbortPolicy();
    /**
     * shutdown and shutdownNow权限
     */
    private static final RuntimePermission shutdownPerm =
        new RuntimePermission("modifyThread");

构造器,,支持最少五种参数,最大七中参数的四种构造器:

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue workQueue,
                              ThreadFactory threadFactory) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             threadFactory, defaultHandler);
    }
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue workQueue,
                              RejectedExecutionHandler handler) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), handler);
    }
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

工人,线程池中执行任务的,线程池就是通过这些工人进行工作的,有核心员工(核心线程)和临时工(人手不够的时候,临时创建的,如果空闲时间厂,就会被裁员),

    private final class Worker
        extends AbstractQueuedSynchronizer
        implements Runnable
    {
        private static final long serialVersionUID = 6138294804551838833L;
        // 工人的本质就是个线程
        final Thread thread;
        // 第一件工作任务
        Runnable firstTask;
      volatile long completedTasks;
        /**
         * 构造器
         */
        Worker(Runnable firstTask) {
            setState(-1); // inhibit interrupts until runWorker
            this.firstTask = firstTask;
            this.thread = getThreadFactory().newThread(this);
        }
        /** 工作  */
        public void run() {
            runWorker(this);
        }
        protected boolean isHeldExclusively() {
            return getState() != 0;
        }
        protected boolean tryAcquire(int unused) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }
        protected boolean tryRelease(int unused) {
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }
        public void lock()        { acquire(1); }
        public boolean tryLock()  { return tryAcquire(1); }
        public void unlock()      { release(1); }
        public boolean isLocked() { return isHeldExclusively(); }
        void interruptIfStarted() {
            Thread t;
            if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                }
            }
        }
    }

核心方法,通过线程池执行任务(这也是线程池的运行原理):

  • 检验任务

  • 获取当前线程池状态

  • 判断上班工人数量是否小于核心员工数

  • 如果小于则招人,安排工作

  • 不小于则判断等候区任务是否排满

  • 如果没有排满则任务排入等候区

  • 如果排满,看是否允许招人,允许招人则招临时工

  • 如果都不行,该线程池无法接收新任务,开始按老板约定的拒绝策略,执行拒绝策略

    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);
        }
        else if (!addWorker(command, false))
            reject(command);
    }

submit()方法是其抽象父类定义的,这里我们就可以明显看到submit与execute的区别,通过submit调用,我们会创建RunnableFuture,并且会返回Future,这里我们可以将返回值类型,告知submit方法,它就会通过泛型约束返回值。

public abstract class AbstractExecutorService implements ExecutorService {
    public Future submit(Runnable task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
    }
    public  Future submit(Runnable task, T result) {
        if (task == null) throw new NullPointerException();
        RunnableFuture ftask = newTaskFor(task, result);
        execute(ftask);
        return ftask;
    }
    public  Future submit(Callable task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }
    ...
}

addWorker()是招人的一个方法:

    private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            // 判断状态,及任务列表
            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 {
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    int rs = runStateOf(ctl.get());
                    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) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

获取任务的方法:

    private Runnable getTask() {
        boolean timedOut = false; // Did the last poll() time out?
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                decrementWorkerCount();
                return null;
            }
            int wc = workerCountOf(c);
            // Are workers subject to culling?
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) {
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
            try {
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }

让员工干活的方法,分配任务,运行任务:

   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);
        }
    }

到此,关于“java线程池的实现原理源码分析”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注编程笔记网站,小编会继续努力为大家带来更多实用的文章!


推荐阅读
  • 本文深入解析了Java 8并发编程中的`AtomicInteger`类,详细探讨了其源码实现和应用场景。`AtomicInteger`通过硬件级别的原子操作,确保了整型变量在多线程环境下的安全性和高效性,避免了传统加锁方式带来的性能开销。文章不仅剖析了`AtomicInteger`的内部机制,还结合实际案例展示了其在并发编程中的优势和使用技巧。 ... [详细]
  • 如何利用Java 5 Executor框架高效构建和管理线程池
    Java 5 引入了 Executor 框架,为开发人员提供了一种高效管理和构建线程池的方法。该框架通过将任务提交与任务执行分离,简化了多线程编程的复杂性。利用 Executor 框架,开发人员可以更灵活地控制线程的创建、分配和管理,从而提高服务器端应用的性能和响应能力。此外,该框架还提供了多种线程池实现,如固定线程池、缓存线程池和单线程池,以适应不同的应用场景和需求。 ... [详细]
  • Ihavetwomethodsofgeneratingmdistinctrandomnumbersintherange[0..n-1]我有两种方法在范围[0.n-1]中生 ... [详细]
  • MySQL Decimal 类型的最大值解析及其在数据处理中的应用艺术
    在关系型数据库中,表的设计与SQL语句的编写对性能的影响至关重要,甚至可占到90%以上。本文将重点探讨MySQL中Decimal类型的最大值及其在数据处理中的应用技巧,通过实例分析和优化建议,帮助读者深入理解并掌握这一重要知识点。 ... [详细]
  • 本文是Java并发编程系列的开篇之作,将详细解析Java 1.5及以上版本中提供的并发工具。文章假设读者已经具备同步和易失性关键字的基本知识,重点介绍信号量机制的内部工作原理及其在实际开发中的应用。 ... [详细]
  • 本文深入解析了JDK 8中HashMap的源代码,重点探讨了put方法的工作机制及其内部参数的设定原理。HashMap允许键和值为null,但键为null的情况只能出现一次,因为null键在内部通过索引0进行存储。文章详细分析了capacity(容量)、size(大小)、loadFactor(加载因子)以及红黑树转换阈值的设定原则,帮助读者更好地理解HashMap的高效实现和性能优化策略。 ... [详细]
  • 本文深入探讨了NoSQL数据库的四大主要类型:键值对存储、文档存储、列式存储和图数据库。NoSQL(Not Only SQL)是指一系列非关系型数据库系统,它们不依赖于固定模式的数据存储方式,能够灵活处理大规模、高并发的数据需求。键值对存储适用于简单的数据结构;文档存储支持复杂的数据对象;列式存储优化了大数据量的读写性能;而图数据库则擅长处理复杂的关系网络。每种类型的NoSQL数据库都有其独特的优势和应用场景,本文将详细分析它们的特点及应用实例。 ... [详细]
  • 在深入掌握Spring框架的事务管理之前,了解其背后的数据库事务基础至关重要。Spring的事务管理功能虽然强大且灵活,但其核心依赖于数据库自身的事务处理机制。因此,熟悉数据库事务的基本概念和特性是必不可少的。这包括事务的ACID属性、隔离级别以及常见的事务管理策略等。通过这些基础知识的学习,可以更好地理解和应用Spring中的事务管理配置。 ... [详细]
  • 尽管我们尽最大努力,任何软件开发过程中都难免会出现缺陷。为了更有效地提升对支持部门的协助与支撑,本文探讨了多种策略和最佳实践,旨在通过改进沟通、增强培训和支持流程来减少这些缺陷的影响,并提高整体服务质量和客户满意度。 ... [详细]
  • MySQL数据库安装图文教程
    本文详细介绍了MySQL数据库的安装步骤。首先,用户需要打开已下载的MySQL安装文件,例如 `mysql-5.5.40-win32.msi`,并双击运行。接下来,在安装向导中选择安装类型,通常推荐选择“典型”安装选项,以确保大多数常用功能都能被正确安装。此外,文章还提供了详细的图文说明,帮助用户顺利完成整个安装过程,确保数据库系统能够稳定运行。 ... [详细]
  • Netty框架中运用Protobuf实现高效通信协议
    在Netty框架中,通过引入Protobuf来实现高效的通信协议。为了使用Protobuf,需要先准备好环境,包括下载并安装Protobuf的代码生成器`protoc`以及相应的源码包。具体资源可从官方下载页面获取,确保版本兼容性以充分发挥其性能优势。此外,配置好开发环境后,可以通过定义`.proto`文件来自动生成Java类,从而简化数据序列化和反序列化的操作,提高通信效率。 ... [详细]
  • 本文总结了一些开发中常见的问题及其解决方案,包括特性过滤器的使用、NuGet程序集版本冲突、线程存储、溢出检查、ThreadPool的最大线程数设置、Redis使用中的问题以及Task.Result和Task.GetAwaiter().GetResult()的区别。 ... [详细]
  • 本文介绍了如何利用 `matplotlib` 库中的 `FuncAnimation` 类将 Python 中的动态图像保存为视频文件。通过详细解释 `FuncAnimation` 类的参数和方法,文章提供了多种实用技巧,帮助用户高效地生成高质量的动态图像视频。此外,还探讨了不同视频编码器的选择及其对输出文件质量的影响,为读者提供了全面的技术指导。 ... [详细]
  • 在CentOS 7环境中安装配置Redis及使用Redis Desktop Manager连接时的注意事项与技巧
    在 CentOS 7 环境中安装和配置 Redis 时,需要注意一些关键步骤和最佳实践。本文详细介绍了从安装 Redis 到配置其基本参数的全过程,并提供了使用 Redis Desktop Manager 连接 Redis 服务器的技巧和注意事项。此外,还探讨了如何优化性能和确保数据安全,帮助用户在生产环境中高效地管理和使用 Redis。 ... [详细]
  • 在软件开发过程中,经常需要将多个项目或模块进行集成和调试,尤其是当项目依赖于第三方开源库(如Cordova、CocoaPods)时。本文介绍了如何在Xcode中高效地进行多项目联合调试,分享了一些实用的技巧和最佳实践,帮助开发者解决常见的调试难题,提高开发效率。 ... [详细]
author-avatar
Life一切安好
这个家伙很懒,什么也没留下!
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有