cheduleExecutorService提交死循环任务

简介: 本文探讨了向调度线程池提交死循环任务时的运行机制。通过分析`scheduleAtFixedRate`方法的调用链路,详细解析了任务从包装为`ScheduledFutureTask`对象到最终执行的过程。核心步骤包括:任务被添加至延时队列、线程池启动Worker执行任务、周期性任务的时间计算与重新提交。若提交的是死循环任务,它将永久占用一个线程,可能导致其他任务无法调度。文章还深入剖析了`getTask`方法中任务获取逻辑及延时计算原理,帮助理解线程池内部工作机制。

       首先提出一个问题,如果向一个调度线程池提交一个死循环任务会发生什么?为了内容的完整性,本文会提到一些在上面列出的文章中已经涉及到的内容。
       比如我们运行下面的代码:

       loopRunner里面只有一个死循环什么也不做,当然这是极端情况,更为一般的情况为在for(;;)里面做一些某种驱动类型的工作,比如Netty的EventLoop一样,那样的循环更有意义,但是本文只是为了学习当向一个调度线程池提交了一个死循环任务之后的运行情况。
下面我们就分析一下scheduleAtFixedRate方法的调用链路:
1、将loopRunner包装成一个ScheduledFutureTask对象,ScheduledFutureTask这个类对于调度线程池至关重要
2、再次包装变为RunnableScheduledFuture对象
3、delayedExecute方法运行,确保任务被正确处理,如果线程池已经被关闭了,那么拒绝任务的提交,否则将任务添加到一个延时队列(workQueue)中去,这是一个具有延时功能的阻塞队列,初始容量为16,每次扩容增加50%的容量,最大容量为Integer.MAX_VALUE
4、运行方法ensurePrestart,确保线程池已经开始工作了,如果线程池里面的线程数量还没有达到设定的corePoolSize,那么就添加一个新的Worker,然后让这个Worker去延时队列去获取任务来执行
5、方法addWorker执行,添加一个Worker,然后让他执行我们提交的任务,下面摘取一段addWorker的方法内容:

/**
 * 完整代码见源码,下面只是摘取了部分,去除了一些不影响阅读的部分
 */
 private boolean addWorker(Runnable firstTask, boolean core) {
   

     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); //添加一个新的worker
                     int s = workers.size();
                     if (s > largestPoolSize)
                         largestPoolSize = s; 
                     workerAdded = true;
                 }
             } finally {
   
                 mainLock.unlock();
             }
             if (workerAdded) {
    //如果添加新的Worker成功,那么就启动它来执行我们提交的任务
                 t.start();
                 workerStarted = true;
             }
         }
     } 
     return workerStarted;
 }

6、第五步中最为重要的一句话就是t.start(),这句话的执行会发生什么?首先看这个t是什么东西:

Worker(Runnable firstTask) {
   
    setState(-1); // inhibit interrupts until runWorker
    this.firstTask = firstTask;
    this.thread = getThreadFactory().newThread(this);
}

       而this就是Worker自身,而Worker是实现了Runnable的,也就是说,t.start()这句话会执行worker自身的run方法

private final class Worker extends AbstractQueuedSynchronizer implements Runnable

7、我们已经知道现在会执行Worker的run方法,下面是run方法的内容:

public void run() {
   
    runWorker(this);
}
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);
    }
}

       首先从Worker中获取其负责的task,如果task为空,那么就去延时队列获取任务,如果没有获取到任务那么线程就可以休息了,如果获取到,那么继续执行下面的内容。主要的就是一句:task.run(),那这句话会发生什么呢?

8、想要知道task.run()之后会发生什么,就需要知道task是个什么东西,第二步的时候说过,也就是我们的任务,只是被包装成了一个RunnableScheduledFuture对象,那现在就去看RunnableScheduledFuture这个方法里面的run会发生什么,下面展示了其run方法的具体细节:

public void run() {
   
    boolean periodic = isPeriodic();
    if (!canRunInCurrentRunState(periodic))
        cancel(false);
    else if (!periodic)
        ScheduledFutureTask.super.run();
    else if (ScheduledFutureTask.super.runAndReset()) {
   
        setNextRunTime();
        reExecutePeriodic(outerTask);
    }
}

       如果这不是一个周期性的任务,那么就执行super的run方法,否则执行runAndReset方法,介于本文是问题导向的文章,所以在此不对super的run方法和runAndReset方法做分析,只要知道这就是执行我们实际提交的任务就好了。也就是说,走到这一步,我们的任务开始运行起来了,也就是我们的那个loopRunner开始无限循环了,下面的代码将永远得不到执行。所以,到这一步就可以解决问题了,向一个调度线程池提交一个死循环的任务,那么这个任务会霸占一个线程一直不会释放,如果很不幸线程池里面只允许有一个线程的话,那么其他提交的任务都将得不到调度执行。

9、为了走通整个流程,我们假设我们提交的不是一个死循环任务,那么提交的任务总是会被执行完的,线程总是会被释放的,那么就会执行setNextRunTime这个方法,下面是这个方法的细节:

private void setNextRunTime() {
   
    long p = period;
    if (p > 0)
        time += p;
    else
        time = triggerTime(-p);
}

       p > 0代表的是scheduleAtFixedRate,p < 0代表的是scheduleWithFixedDelay,两者的区别在于前者总是按照设定的轨迹来设定下次应该调度的时间,而后者总是在任务执行完成之后再根据周期设定下一次应该执行的时间。我们只分析前者。对于第一次提交的任务,time等于当前时间 + 首次延时执行的时间,对于delay等于0的情况下,首次提交任务的time就是当前时间,然后 + p代表的是下一次应该被调度的时间。

10、我们发现,每个任务都是在执行完一次之后再设定下次执行任务的时间的,这也特别关键。设定好下次调度的时间,那么就要开始去准备执行它吧,也就是reExecutePeriodic方法会执行,下面是reExecutePeriodic方法的内容:

void reExecutePeriodic(RunnableScheduledFuture<?> task) {
   
    if (canRunInCurrentRunState(true)) {
   
        super.getQueue().add(task);
        if (!canRunInCurrentRunState(true) && remove(task))
            task.cancel(false);
        else
            ensurePrestart();
    }
}

       这个方法只是将任务重新提交到了延时队列而已,一次完整的流程到底也就结束了,为了内容的完整性,再来分析一下一个Worker从延时队列获取任务时的情况。回到第七步,我们有一个方法没有提到,那就是getTask():

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

       我们主要来看两个方法: poll/take,这两个方法都是从延时队列获取一个任务,下面是poll的代码,take会阻塞一直到获取到内容,而poll则不会阻塞:

public RunnableScheduledFuture<?> poll(long timeout, TimeUnit unit) throws InterruptedException {
   
    long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
   
        for (;;) {
   
            RunnableScheduledFuture<?> first = queue[0];
            if (first == null) {
   
                if (nanos <= 0)
                    return null;
                else
                    nanos = available.awaitNanos(nanos);
            } else {
   
                long delay = first.getDelay(NANOSECONDS);
                if (delay <= 0)
                    return finishPoll(first);
                if (nanos <= 0)
                    return null;
                first = null; // don't retain ref while waiting
                if (nanos < delay || leader != null)
                    nanos = available.awaitNanos(nanos);
                else {
   
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
   
                        long timeLeft = available.awaitNanos(delay);
                        nanos -= delay - timeLeft;
                    } finally {
   
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
   
        if (leader == null && queue[0] != null)
            available.signal();
        lock.unlock();
    }
}

       poll的代码最为核心的内容就是,获取队列首部的任务,然后获取其延时时间,这个时间是我们在完成一次调度之后设置的下次调度时间,如果任务的运行时间大于我们设定的周期的话,这个延时时间就是负数的,那么就会被立即执行,否则会等到设定的时间,时间到了再返回给Worker执行。
       最后把getDelay方法的细节粘出来,这样内容就完整了,其中的time是我们设定的:

public long getDelay(TimeUnit unit) {
   
    return unit.convert(time - now(), NANOSECONDS);
}
相关文章
|
Java Maven 索引
Logback:同时按照日期和大小分割日志(最新日志可以不带日期或数字)
Logback:同时按照日期和大小分割日志(最新日志可以不带日期或数字)
Logback:同时按照日期和大小分割日志(最新日志可以不带日期或数字)
|
缓存 运维 监控
10分钟带你了解 Linux 系统中的 Top 命令
`top`命令是Linux系统中用于实时监控系统资源利用率的工具,展示CPU、内存使用情况及进程状态。启动`top`只需在终端输入`top`。默认按CPU使用率排序,可通过`P`、`M`、`T`键改变排序。使用`k`键可结束进程,`d`键调整刷新率,`q`键退出。输出信息包括系统负载、进程状态、内存使用等。通过进程列表,可以观察到每个进程的CPU和内存占用、用户、运行时间等。了解`top`能帮助测试工程师排查性能问题。
|
Swift iOS开发 Ruby
iOS CocoaPods 使用以及常见问题(上)
iOS CocoaPods 使用以及常见问题
1141 0
POST 请求出现异常!java.io.IOException: Server returned HTTP response code: 400 for URL
POST 请求出现异常!java.io.IOException: Server returned HTTP response code: 400 for URL
5274 0
|
人工智能 Java
Java 中数组Array和列表List的转换
本文介绍了数组与列表之间的相互转换方法,主要包括三部分:1)使用`Collections.addAll()`方法将数组转为列表,适用于引用类型,效率较高;2)通过`new ArrayList&lt;&gt;()`构造器结合`Arrays.asList()`实现类似功能;3)利用JDK8的`Stream`流式计算,支持基本数据类型数组的转换。此外,还详细讲解了列表转数组的方法,如借助`Stream`实现不同类型数组间的转换,并附带代码示例与执行结果,帮助读者深入理解两种数据结构的互转技巧。
1166 1
Java 中数组Array和列表List的转换
|
Java Spring
使用 Spring Boot 多个定时任务阻塞问题的解决方案
我是小假 期待与你的下一次相遇 ~
715 5
可能是由于IntelliJ IDEA的某些设置或者插件冲突导致的
可能是由于IntelliJ IDEA的某些设置或者插件冲突导致的
2446 1
|
NoSQL 安全 MongoDB
MongoDB 数据精简指南:删除文档操作详解
MongoDB 数据精简指南:删除文档操作详解
795 0
|
存储 NoSQL JavaScript
mongodb时间处理,时区处理
mongodb时间处理,时区处理
4708 2