spring定时任务详解

使用

很方便的使用方法是用@Scheduled注解，在Spring配置文件中加入相应的配置。

其中ThreadPoolTaskScheduler的poolSize属性代表用于执行定时任务的线程数。例如有两个定时任务触发的时间相同，如果只有一个线程，那么有一个任务需要等到另一个任务执行完了才能执行，如果线程数是2则可以并行执行。

<task:annotation-driven executor="asyncTaskExecutor" scheduler="scheduledTaskExecutor"/>

<bean id="scheduledTaskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler">
<property name="poolSize" value="${scheduledTaskExecutor.poolSize}"/>
</bean>

源码解读

ScheduledAnnotationBeanPostProcessor

这个类会解析@Schedule注解的方法，获取方法检查是否没有参数，然后包装成ScheduledMethodRunnable，根据时间的配置是cron还是fixedDelay还是fixedRate，放进这个processor的一个Tasks的map中，这是三个以Rannable为key，时间配置为value的map.

    private final Map<Runnable, String> cronTasks = new HashMap();
private final Map<Runnable, Long> fixedDelayTasks = new HashMap();
private final Map<Runnable, Long> fixedRateTasks = new HashMap();

这个类实现了ApplicationListener<ContextRefreshedEvent> ，当有event发布时，如果上文提到的三个map至少有一个不为空，则新建一个ScheduledTaskRegistrar，把三个map set进去，scheduler也set进去（这个scheduler应该就是在task:annotation-driven配置的）。

如果没有配置scheduler则会取applicationContext中所有的type是TaskScheduler和ScheduledExecutorService的bean（只能有一个，否则会报错）。

最后调用registrar.afterPropertiesSet() 实际是把上文中三个map里的task作为参数调用taskScheduler.schedule(），结果放入private final Set<ScheduledFuture<?>> scheduledFutures = new LinkedHashSet();中。

ThreadPoolTaskScheduler

Spring的ThreadPoolTaskScheduler类的hierarchy结构如图
spring定时任务详解
这个对象在初始化时会生成一个ScheduledThreadPoolExecutor作为schedulerExecutor，这个事cunrrent包中的类。核心线程池大小就是我们配置的poolSize属性，最大线程池大小是Integer.MAX_VALUE,keepAliveTime为0,队列是DelayedWorkQueue，这个队列有一个属性private final DelayQueue<RunnableScheduledFuture> dq = new DelayQueue<RunnableScheduledFuture>(); 对这个队列的操作实际是是对这个DelayQueue的操作，这个队列大小是Integer.MAX_VALUE。

schedule方法实际是调用的ScheduledThreadPoolExecutor对象的schedule方法。
非周期性任务：

 public ScheduledFuture schedule(Runnable task, Date startTime) {
        ScheduledExecutorService executor = this.getScheduledExecutor();
long initialDelay = startTime.getTime() - System.currentTimeMillis();

try {
return executor.schedule(this.errorHandlingTask(task, false), initialDelay, TimeUnit.MILLISECONDS);
        } catch (RejectedExecutionException var7) {
throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, var7);
        }
    }

周期性任务：

public ScheduledFuture scheduleAtFixedRate(Runnable task, Date startTime, long period) {
        ScheduledExecutorService executor = this.getScheduledExecutor();
long initialDelay = startTime.getTime() - System.currentTimeMillis();

try {
return executor.scheduleAtFixedRate(this.errorHandlingTask(task, true), initialDelay, period, TimeUnit.MILLISECONDS);
        } catch (RejectedExecutionException var9) {
throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, var9);
        }
    }

this.errorHandlingTask(task, true)这里把任务包装成了DelegatingErrorHandlingRunnable代理对象，在执行出错时会执行ErrorHanler的handleError方法。方法里的true和false代表是不是周期性任务。

下面再来仔细看ScheduledThreadPoolExecutor的schedule方法。

public ScheduledFuture<?> schedule(Runnable command,
long delay,
                                       TimeUnit unit) {
if (command == null || unit == null)
throw new NullPointerException();
if (delay < 0) delay = 0;
long triggerTime = now() + unit.toNanos(delay);
        RunnableScheduledFuture<?> t = decorateTask(command,
new ScheduledFutureTask<Boolean>(command, null, triggerTime));
        delayedExecute(t);
return t;
    }

把task包装成ScheduledFutureTask的对象之后执行delayedExecute方法。

dealyedExecute方法如下，其实就是把任务放进阻塞队列中：

 private void delayedExecute(Runnable command) {
if (isShutdown()) {
            reject(command);
return;
        }
// Prestart a thread if necessary. We cannot prestart it
// running the task because the task (probably) shouldn't be
// run yet, so thread will just idle until delay elapses.
if (getPoolSize() < getCorePoolSize())
            prestartCoreThread();

super.getQueue().add(command);
    }

ScheduledFutureTask类是ScheduledThreadPoolExecutor的内部类,实现了delayed接口。属性有：

  /** Sequence number to break ties FIFO */
private final long sequenceNumber;
/** The time the task is enabled to execute in nanoTime units */
private long time;
/**
 * Period in nanoseconds for repeating tasks. A positive
 * value indicates fixed-rate execution. A negative value
 * indicates fixed-delay execution. A value of 0 indicates a
 * non-repeating task.
 */
private final long period;

在队列中根据time排序,如果time相同则根据sequence排序(加入顺序).有属性time为执行时间，period为执行周期，不为0表示周期性执行。run方法中，判断如果不是周期性的任务则直接执行callable的run方法（这个就是我们自己写的带有@scheduled注解的方法），如果是周期性的任务则执行完之后加上delay的时间再次把自己放入阻塞队里中

在ThreadPoolExcecutor中，worker的run方法会take（阻塞）出delayQueue中的任务，然后调用任务的run方法。如果排名最前的任务时间还没到返回null，这样任务不会被执行。

 public E poll() {
final ReentrantLock lock = this.lock;
        lock.lock();
try {
            E first = q.peek();
if (first == null || first.getDelay(TimeUnit.NANOSECONDS) > 0)
return null;
else {
                E x = q.poll();
assert x != null;
if (q.size() != 0)
                    available.signalAll();
return x;
            }
        } finally {
            lock.unlock();
        }
    }

在ThreadPoolExecutor中，addThread方法增加一个worker,启动线程，这样worker就一直去拿队列中的任务来执行。

private Thread addThread(Runnable firstTask) {
        Worker w = new Worker(firstTask);
        Thread t = threadFactory.newThread(w); //worker作为runnable传入Thread，线程start之后会执行worker的run方法
if (t != null) {
            w.thread = t;
            workers.add(w);
int nt = ++poolSize;
if (nt > largestPoolSize)
                largestPoolSize = nt;
        }
return t;
    }

秒客网

spring定时任务详解

使用

源码解读

ScheduledAnnotationBeanPostProcessor

ThreadPoolTaskScheduler

相关文章