ScheduledThreadPoolExecutor的相关知识

ThreadPool

A simple C++11 Thread Pool implementation

项目地址：https://gitcode.com/gh_mirrors/th/ThreadPool

免费下载资源

IT界的吉祥物

27400人浏览 · 2017-10-19 16:36:05

IT界的吉祥物 · 2017-10-19 16:36:05 发布

转载请标明出处:【顾林海的博客】

个人开发的微信小程序，目前功能是书籍推荐，后续会完善一些新功能，希望大家多多支持！
在这里插入图片描述

##前言
ScheduledThreadPoolExecutor继承自ThreadPoolExecutor，而ThreadPoolExecutor是线程池的核心实现类，用来执行被提交的任务，ScheduledThreadPoolExecutor是一个实现类，可以在给定的延迟后运行命令，或者定期执行命令。

##ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor定义了四个构造函数，这四个构造函数如下：

/**
* @param corePoolSize 核心线程池的大小
*/
public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE,
            DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
            new DelayedWorkQueue());
}

/**
* @param corePoolSize  核心线程池的大小
* @param threadFactory 用于设置创建线程的工厂
*/
public ScheduledThreadPoolExecutor(int corePoolSize,
                                   ThreadFactory threadFactory) {
    super(corePoolSize, Integer.MAX_VALUE,
            DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
            new DelayedWorkQueue(), threadFactory);
}

/**
* @param corePoolSize 核心线程池的大小
* @param handler      饱和策略
*/
public ScheduledThreadPoolExecutor(int corePoolSize,
                                   RejectedExecutionHandler handler) {
    super(corePoolSize, Integer.MAX_VALUE,
            DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
            new DelayedWorkQueue(), handler);
}

/**
* @param corePoolSize  核心线程池的大小
* @param threadFactory 用于设置创建线程的工厂
* @param handler       饱和策略
*/
public ScheduledThreadPoolExecutor(int corePoolSize,
                                   ThreadFactory threadFactory,
                                   RejectedExecutionHandler handler) {
    super(corePoolSize, Integer.MAX_VALUE,
            DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
            new DelayedWorkQueue(), threadFactory, handler);
}

通过源码可以发现，ScheduledThreadPoolExecutor的构造器都是调用父类的构造器也就是ThreadPoolExecutor的构造器，以此来创建一个线程池。

ThreadPoolExecutor的构造器如下：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
            Executors.defaultThreadFactory(), defaultHandler);
}


public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
            threadFactory, defaultHandler);
}

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          RejectedExecutionHandler handler) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
            Executors.defaultThreadFactory(), handler);
}

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> 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;
}

创建一个线程池时需要输入几个参数，如下：

corePoolSize（线程池的基本大小）：当提交一个任务到线程池时，线程池会创建一个线程来执行任务，即使其它空闲的基本线程能够执行新任务也会创建线程，等到需要执行的任务数大于线程池基本大小时就不再创建，会把到达的任务放到缓存队列当中。如果调用了线程池的prestartAllCoreThreads()方法，线程池会提前创建并启动所有基本线程，或调用线程池的prestartCoreThread()方法，线程池会提前创建一个线程。
maximumPoolSize（线程池最大数量）：线程池允许创建的最大线程数。如果队列满了，并且已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务。值得注意的是，如果使用了无界的任务队列这个参数就没什么效果。
KeepAliveTime（线程活动保持时间）：线程池的工作线程空闲后，保持存货的时间。如果任务很多，并且每个任务执行的时间比较短，可以调大时间，提供线程的利用率。
unit（线程活动保持时间的单位）：可选的单位有天（DAYS）、小时（HOURS）、分钟（MINUTES）、毫秒（MILLISECONDS）、微妙（MICROSECONDS）、千分之一毫秒和纳秒（NANOSECONDS、千分之一微妙）。
workQueue（任务队列）：用于保持等待执行的任务的阻塞队列，可以选择以下几个阻塞队列。
（1）ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按FIFO（先进先出）原则对元素进行排序。
（2）LinkedBlockingQueue：一个基于链表结构的阻塞队列，此队列按FIFO排序元素，吞吐量通常要高于ArrayBlockingQueue。
（3）SynchronousQueue：一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue。
（4）PriorityBlockingQueue：一个具有优先级的无限阻塞队列。
ThreadFactory：用于设置创建线程的工厂，可以通过线程工厂给每个创建出来的线程设置更有意义的名字。
RejectedExecutionHandler（饱和策略）：当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是AbortPolicy，表示无法处理新任务时抛出异常。在JDK1.5中Java线程池框架提供了4种策略（也可通过实现RejectedExecutionHandler接口自定义策略）。
（1）AbortPolicy：直接抛出异常。
（2）CallerRunsPolicy：只用调用者所在线程来运行任务。
（3）DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务。
（4）DiscardPolicy：处理，丢弃掉。

在ScheduledThreadPoolExecutor构造器中使用了工作队列java.util.concurrent.ScheduledThreadPoolExecutor.DelayedWorkQueue，DelayedWorkQueue是一个无界的BlockingQueue，
用于放置实现了Delayed接口的对象，其中的对象只能在其到期才能从队列中取走。

由于ScheduledThreadPoolExecutor继承自ThreadPoolExecutor，因此它也实现了ThreadPoolExecutor的方法，如下：

这里写图片描述

同时它也有自己的定时执行任务的方法：

schedule (Runnable task, long delay, TimeUnit timeunit)
schedule (Callable task, long delay, TimeUnit timeunit)
scheduleAtFixedRate (Runnable, long initialDelay, long period, TimeUnit timeunit)
scheduleWithFixedDelay (Runnable, long initialDelay, long period, TimeUnit timeunit)

方法说明：

方法1：延迟delay时间后开始执行task，无法获取task的执行结果。

方法2： 延迟delay时间后开始执行callable，它接收的是一个Callable实例，此方法会返回一个ScheduleFuture对象，通过ScheduleFuture我们可以取消一个未执行的task，也可以获得这个task的执行结果。

方法3：延迟initialDelay时间后开始执行command，并且按照period时间周期性重复调用，当任务执行时间大于间隔时间时，之后的任务都会延迟，此时与Timer中的schedule方法类似。

方法4延迟initialDelay时间后开始执行command，并且按照period时间周期性重复调用，这里的间隔时间delay是等上一个任务完全执行完毕才开始计算。

ScheduledThreadPoolExecutor把待调度的任务放到一个DelayedWorkQueue ，并且DelayedWorkQueue 是一个无界队列，ThreadPoolExecutor的maximumPoolSize在ScheduledThreadPoolExecutor中没有什么意义。整个ScheduledThreadPoolExecutor的运行可以分为两大部分。

1、当调用ScheduledThreadPoolExecutor的scheduleAtFixedRate()方法或者scheduleWithFixedDelay()方法时，会向ScheduledThreadPoolExecutor的DelayedWorkQueue 添加一个实现了RunnableScheduleFuture接口的ScheduledFutureTask，如下ScheduledThreadPoolExecutor其中的一个schedule方法。

public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                       long delay,
                                       TimeUnit unit) {
    if (callable == null || unit == null)
        throw new NullPointerException();
    RunnableScheduledFuture<V> t = decorateTask(callable,
        new ScheduledFutureTask<V>(callable,
                                   triggerTime(delay, unit),
                                   sequencer.getAndIncrement()));
    delayedExecute(t);
    return t;
}

private void delayedExecute(RunnableScheduledFuture<?> task) {
    if (isShutdown())
        reject(task);
    else {
        super.getQueue().add(task);//向ScheduledThreadPoolExecutor的DelayedWorkQueue添加一个实现了RunnableScheduleFuture接口的ScheduledFutureTask
        if (isShutdown() &&
            !canRunInCurrentRunState(task.isPeriodic()) &&
            remove(task))
            task.cancel(false);
        else
            ensurePrestart();
    }
}

2、线程池中的线程从DelayedWorkQueue 中获取ScheduledFutureTask，然后执行。

ScheduledFutureTask是ScheduledThreadPoolExecutor的内部类并继承自FutureTask，包含3个成员变量。

private final long sequenceNumber;

private volatile long time;

private final long period;

long型成员变量sequenceNumber，表示这个任务被添加到ScheduledThreadPoolExecutor中的序号。
long型成员变量time，表示这个任务将要被执行的具体时间。
long型成员变量period，表示任务执行的间隔周期。

ScheduledFutureTask内部实现了compareTo()方法，用于对task的排序。

public int compareTo(Delayed other) {
    if (other == this) // compare zero if same object
        return 0;
    if (other instanceof ScheduledFutureTask) {
        ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
        long diff = time - x.time;
        if (diff < 0)
            return -1;
        else if (diff > 0)
            return 1;
        else if (sequenceNumber < x.sequenceNumber)
            return -1;
        else
            return 1;
    }
    long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
    return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
}

排序时，time小的排在前面，如果两个ScheduledFutureTask的time相同，就比较sequenceNumber，sequenceNumber小的排在前面。

DelayedWorkQueue 内部使用了二叉堆算法，DelayedWorkQueue 中的元素第一个元素永远是延迟时间最小的那个元素。当执行 schedule 方法是。如果不是重复的任务，那任务从 DelayedWorkQueue 取出之后执行完了就结束了。如果是重复的任务，那在执行结束前会重置执行时间并将自己重新加入到 DelayedWorkQueue 中

总结来说，ScheduledThreadPoolExecutor是一个实现ScheduledExecutorService的可以调度任务的执行框架；DelayedWorkQueue是一个数组实现的阻塞队列，根据任务所提供的时间参数来调整位置，实际上就是个小根堆（优先队列）；ScheduledFutureTask包含任务单元，存有时间、周期、外部任务、堆下标等调度过程中必须用到的参数，被工作线程执行。ScheduledThreadPoolExecutor与Timer都是用作定时任务，它们直接的差异是Timer使用的是绝对时间，系统时间的改变会对Timer产生一定的影响；而ScheduledThreadPoolExecutor使用的是相对时间，不会导致这个问题。Timer使用的是单线程来处理任务，长时间运行的任务会导致其他任务的延迟处理；而ScheduledThreadPoolExecutor可以自定义线程数量。并且Timer没有对运行时异常进行处理，一旦某个任务触发运行时异常，会导致整个Timer崩溃；而ScheduledThreadPoolExecutor对运行时异常做了捕获（通过afterExecute()回调方法中进行处理），所以更安全。

GitHub 加速计划 / th / ThreadPool

下载

A simple C++11 Thread Pool implementation

最近提交(Master分支：3 个月前 )

9a42ec13 - 10 年前

fcc91415 - 10 年前

GitCode 开源社区

旨在为数千万中国开发者提供一个无缝且高效的云端环境，以支持学习、使用和贡献开源项目。

更多推荐

[转载]在Windows环境下安装GNU Radio

转自：在Windows环境下安装GNURadio_恐弱智_新浪博客GNU Radio是用Python开发的，大部分开源的工程能够在Linux环境下运行良好，而Windows下却运行的很勉强，而且安装配置都很复杂。GNU Radio算是个例外了，不光提供了Windows的二进制安装，还有比较详细的说明。我是Python小白，所以折腾了好久才弄好，特意记录下来，免得以后再装还折腾。GNU Radio的

GitCode 开源社区

centOS 8 使用dnf安装Docker

DNF是什么？CentOS 8使用YUM软件包管理器版本v4.0.4。现在，该版本使用DNF(已删除YUM)。DNF是软件包管理器。它会在Linux发行版上安装，执行更新并删除软件包。使用DNF安装Docker跳过具有损坏依赖性的程序包一个有效的解决方案是使您的CentOS 8系统使用以下--nobest命令安装最符合条件的版本：sudo dnf install docker...

GitCode 开源社区

定时同步数据库表(mysql+linux+crontab)

sync.sh里面的参数需要改变，ip/username/password/database/tablesync.sh#!/bin/sh# Please change the IP and password of the data source db.# Then change the table name.filename=/home/nington/db/$(date +%Y-%m