JAVA 分析线程池中的keepAliveTime参数具体实现

简介: JAVA 分析线程池中的keepAliveTime参数具体实现

分析线程池中的keepAliveTime参数具体实现

参数意义

创建线程池时,有一个重要参数就是keepAliveTime,标记线程空闲多久后被释放。

那么他到底是怎么实现的呢?

猜想

有一个线程在维护时间,可笑。专门有一个线程去维护,浪费资源,而且时间也不够精确。而且还要开辟空间记录线程开始空闲的时间,消耗空间。

源码分析

java.util.concurrent.ThreadPoolExecutor#getTask

/**
     * Performs blocking or timed wait for a task, depending on
     * current configuration settings, or returns null if this worker
     * must exit because of any of:
     * 1. There are more than maximumPoolSize workers (due to
     *    a call to setMaximumPoolSize).
     * 2. The pool is stopped.
     * 3. The pool is shutdown and the queue is empty.
     * 4. This worker timed out waiting for a task, and timed-out
     *    workers are subject to termination (that is,
     *    {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
     *    both before and after the timed wait, and if the queue is
     *    non-empty, this worker is not the last thread in the pool.
     *
     * @return task, or null if the worker must exit, in which case
     *         workerCount is decremented
     */
    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 {
                // 这一步是关键,需要了解poll和take的区别,take会进行阻塞。
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) {
                timedOut = false;
            }
        }
    }
  • poll 方法作用是移除并返回队列的头节点。但是如果当队列里面是空的,没有任何东西可以移除的时候,便会返回 null 作为提示。
  • 带时间参数的 poll 方法:如果能够移除,便会立刻返回这个节点的内容;如果队列是空的就会进行等待,等待时间正是我们指定的时间,直到超时时间到了,如果队列里依然没有元素可供移除,便会返回 null 作为提示。
  • take 方法的作用是获取并移除队列的头结点。通常在队列里有数据的时候会正常取出数据并删除;但是如果执行 take 的时候队列里无数据,则阻塞,直到队列里有数据;一旦队列里有数据了,就会立刻解除阻塞状态,并且取到数据。
  • timed为true则标志着:允许核心线程超时被释放或者当前线程数超过核心线程数。一旦为tue,就会去使用阻塞对垒的poll方法,如果keepAliveTime的时间里获取不到任务,就会返回Null,在上一级,也就是runWorker方法中去释放资源。

java.util.concurrent.ThreadPoolExecutor#runWorker

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循环获取任务,如果获取不到,就跳出循环,在finally中进行释放。
            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 {
            // 去释放当前的worker w。
            processWorkerExit(w, completedAbruptly);
        }
    }

阻塞队列的poll方法又是怎么实现超时的呢?

但是我们较真来看,打破砂锅问到底,阻塞队列又是如何做到去完成超时的呢?

以java.util.concurrent.DelayQueue为例

源码分析

java.util.concurrent.DelayQueue#poll(long, java.util.concurrent.TimeUnit)

/**
     * Retrieves and removes the head of this queue, waiting if necessary
     * until an element with an expired delay is available on this queue,
     * or the specified wait time expires.
     *
     * @return the head of this queue, or {@code null} if the
     *         specified waiting time elapses before an element with
     *         an expired delay becomes available
     * @throws InterruptedException {@inheritDoc}
     */
    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            for (;;) {
                E first = q.peek();
                if (first == null) {
                    if (nanos <= 0)
                        // 已经超时就返回null
                        return null;
                    else
                        // 未超时就去await
                        nanos = available.awaitNanos(nanos);
                } else {
                    long delay = first.getDelay(NANOSECONDS);
                    if (delay <= 0)
                        return q.poll();
                    if (nanos <= 0)
                        return null;
                    first = null; // don't retain ref while waiting
                    if (nanos < delay || leader != null)
                        // 未超时就去await
                        nanos = available.awaitNanos(nanos);
                    else {
                        Thread thisThread = Thread.currentThread();
                        leader = thisThread;
                        try {
                            // 未超时就去await
                            long timeLeft = available.awaitNanos(delay);
                            nanos -= delay - timeLeft;
                        } finally {
                            if (leader == thisThread)
                                leader = null;
                        }
                    }
                }
            }
        } finally {
            if (leader == null && q.peek() != null)
                available.signal();
            lock.unlock();
        }
    }

可以看到反复调用了available.awaitNanos(delay)方法,继续往下看

java.util.concurrent.locks.AbstractQueuedSynchronizer.ConditionObject#awaitNanos

/**
         * Implements timed condition wait.
         * <ol>
         * <li> If current thread is interrupted, throw InterruptedException.
         * <li> Save lock state returned by {@link #getState}.
         * <li> Invoke {@link #release} with saved state as argument,
         *      throwing IllegalMonitorStateException if it fails.
         * <li> Block until signalled, interrupted, or timed out.
         * <li> Reacquire by invoking specialized version of
         *      {@link #acquire} with saved state as argument.
         * <li> If interrupted while blocked in step 4, throw InterruptedException.
         * </ol>
         */
        public final long awaitNanos(long nanosTimeout)
                throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            final long deadline = System.nanoTime() + nanosTimeout;
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                if (nanosTimeout <= 0L) {
                    transferAfterCancelledWait(node);
                    break;
                }
                if (nanosTimeout >= spinForTimeoutThreshold)
                    // 这一步去做了等待
                    LockSupport.parkNanos(this, nanosTimeout);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
                nanosTimeout = deadline - System.nanoTime();
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
            return deadline - System.nanoTime();
        }

看到执行了LockSupport.parkNanos(this, nanosTimeout);

java.util.concurrent.locks.LockSupport#parkNanos(java.lang.Object, long)

public static void parkNanos(Object blocker, long nanos) {
        if (nanos > 0) {
            Thread t = Thread.currentThread();
            setBlocker(t, blocker);
            UNSAFE.park(false, nanos);
            setBlocker(t, null);
        }
    }

执行了UNSAFE.park(false, nanos);

sun.misc.Unsafe#park

public native void park(boolean var1, long var2);

这是一个native方法,一般来说到了native我就不看了…c++都还给夫子了。

网上搜搜看吧!!

转载分析本地方法的博客:https://www.jianshu.com/p/37ef66eddca6

也不是特别详细,没有到操作系统底层。

个人猜测操作系统是包了一层while循环,在未超时时就放弃这个CPU时间片,让出CPU,等待下次调度再做判断。

总结

不算太虎头蛇尾吧,至少把jdk中是怎么做到实现keepAliveTime参数的源码给看懂了。

欢迎大家交流指出JRE中阻塞队列调用的UNSAFE如何做到超时,以及对应到操作系统如何配合这个超时指令。

wanglh
+关注
目录
打赏
0
0
0
0
1
分享
相关文章
|
5天前
|
【源码】【Java并发】从InheritableThreadLocal和TTL源码的角度来看父子线程传递
本文涉及InheritableThreadLocal和TTL,从源码的角度,分别分析它们是怎么实现父子线程传递的。建议先了解ThreadLocal。
34 4
【源码】【Java并发】从InheritableThreadLocal和TTL源码的角度来看父子线程传递
Minecraft配置文件参数说明(JAVA服务器篇)
Minecraft JAVA版服务器启动后会生成server.properties配置文件,位于minecraft_server/根目录下。该文件包含多项关键设置,如游戏模式(gamemode)、最大玩家数(max-players)、难度(difficulty)等。此文档详细说明了各配置项的功能与默认值,帮助用户高效管理服务器环境。
112 3
Java网络编程,多线程,IO流综合小项目一一ChatBoxes
**项目介绍**:本项目实现了一个基于TCP协议的C/S架构控制台聊天室,支持局域网内多客户端同时聊天。用户需注册并登录,用户名唯一,密码格式为字母开头加纯数字。登录后可实时聊天,服务端负责验证用户信息并转发消息。 **项目亮点**: - **C/S架构**:客户端与服务端通过TCP连接通信。 - **多线程**:采用多线程处理多个客户端的并发请求,确保实时交互。 - **IO流**:使用BufferedReader和BufferedWriter进行数据传输,确保高效稳定的通信。 - **线程安全**:通过同步代码块和锁机制保证共享数据的安全性。
82 23
|
3天前
|
Java 集合框架详解:系统化分析与高级应用
本文深入解析Java集合框架,涵盖List、Set、Map等核心接口及其常见实现类,如ArrayList、HashSet、HashMap等。通过对比不同集合类型的特性与应用场景,帮助开发者选择最优方案。同时介绍Iterator迭代机制、Collections工具类及Stream API等高级功能,提升代码效率与可维护性。适合初学者与进阶开发者系统学习与实践。
19 0
|
3天前
|
java中一个接口A,以及一个实现它的类B,一个A类型的引用对象作为一个方法的参数,这个参数的类型可以是B的类型吗?
本文探讨了面向对象编程中接口与实现类的关系,以及里氏替换原则(LSP)的应用。通过示例代码展示了如何利用多态性将实现类的对象传递给接口类型的参数,满足LSP的要求。LSP确保子类能无缝替换父类或接口,不改变程序行为。接口定义了行为规范,实现类遵循此规范,从而保证了多态性和代码的可维护性。总结来说,接口与实现类的关系天然符合LSP,体现了多态性的核心思想。
11 0
列表结构与树结构转换分析与工具类封装(java版)
本文介绍了将线性列表转换为树形结构的实现方法及工具类封装。核心思路是先获取所有根节点,将其余节点作为子节点,通过递归构建每个根节点的子节点。关键在于节点需包含 `id`、`parentId` 和 `children` 三个属性。文中提供了两种封装方式:一是基于基类 `BaseTree` 的通用工具类,二是使用函数式接口实现更灵活的方式。推荐使用后者,因其避免了继承限制,更具扩展性。代码示例中使用了 Jackson 库进行 JSON 格式化输出,便于结果展示。最后总结指出,理解原理是进一步优化和封装的基础。
|
1月前
|
【源码】【Java并发】【线程池】邀请您从0-1阅读ThreadPoolExecutor源码
当我们创建一个`ThreadPoolExecutor`的时候,你是否会好奇🤔,它到底发生了什么?比如:我传的拒绝策略、线程工厂是啥时候被使用的? 核心线程数是个啥?最大线程数和它又有什么关系?线程池,它是怎么调度,我们传入的线程?...不要着急,小手手点上关注、点赞、收藏。主播马上从源码的角度带你们探索神秘线程池的世界...
135 0
【源码】【Java并发】【线程池】邀请您从0-1阅读ThreadPoolExecutor源码
|
24天前
|
线程池的核心参数有哪些 ?
corePoolSize 核心线程数量 maximumPoolSize 最大线程数量 keepAliveTime 线程保持时间,N个时间单位 unit 时间单位(比如秒,分) workQueue 阻塞队列 threadFactory 线程工厂 handler 线程池拒绝策略
|
2月前
|
【Java并发】【线程池】带你从0-1入门线程池
欢迎来到我的技术博客!我是一名热爱编程的开发者,梦想是编写高端CRUD应用。2025年我正在沉淀中,博客更新速度加快,期待与你一起成长。 线程池是一种复用线程资源的机制,通过预先创建一定数量的线程并管理其生命周期,避免频繁创建/销毁线程带来的性能开销。它解决了线程创建成本高、资源耗尽风险、响应速度慢和任务执行缺乏管理等问题。
188 60
【Java并发】【线程池】带你从0-1入门线程池
AI助理

你好,我是AI助理

可以解答问题、推荐解决方案等