实现你自己的 RunnableScheduledFuture 接口来执行延迟和周期性任务。【回头再看看】

自定义在计划的线程池内运行的任务

计划的线程池是 Executor 框架的基本线程池的扩展，允许你定制一个计划来执行一段时间后需要被执行的任务。 它通过 ScheduledThreadPoolExecutor 类来实现，并允许运行以下这两种任务：

Delayed 任务：这种任务在一段时间后仅执行一次。

Periodic 任务：这种任务在延迟后执行，然后通常周期性运行

Delayed 任务可以执行 Callable 和 Runnable 对象，但是 periodic任务只能执行 Runnable 对象。全部任务通过计划池执行的都必须实现 RunnableScheduledFuture 接口。在这个指南，你将学习如何实现你自己的 RunnableScheduledFuture 接口来执行延迟和周期性任务。

实例代码

package com.packtpub.java7.concurrency.chapter7.recipe05.core;

import java.util.Date;
import java.util.concurrent.TimeUnit;

import com.packtpub.java7.concurrency.chapter7.recipe05.task.MyScheduledThreadPoolExecutor;
import com.packtpub.java7.concurrency.chapter7.recipe05.task.Task;

/**
* Main class of the example. Creates a MyScheduledThreadPoolExecutor and
* executes a delayed task and a periodic task in it.
*/
public class Main {

/**
* @param args
*/
public static void main(String[] args) throws Exception {

/*
* Create a MyScheduledThreadPool object
*/
MyScheduledThreadPoolExecutor executor=new MyScheduledThreadPoolExecutor(2);

/*
* Create a task object
*/
Task task=new Task();

/*
* Write the start date of the execution
*/
System.out.printf("Main: %s\n",new Date());

/*
* Send to the executor a delayed task. It will be executed after 1 second of delay
*/
executor.schedule(task, 1, TimeUnit.SECONDS);

/*
* Sleeps the thread three seconds
*/
TimeUnit.SECONDS.sleep(3);

/*
* Create another task
*/
task=new Task();

/*
* Write the actual date again
*/
System.out.printf("Main: %s\n",new Date());

/*
* Send to the executor a delayed task. It will begin its execution after 1 second of dealy
* and then it will be executed every three seconds
*/
executor.scheduleAtFixedRate(task, 1, 3, TimeUnit.SECONDS);

/*
* Sleep the thread during ten seconds
*/
TimeUnit.SECONDS.sleep(10);

/*
* Shutdown the executor
*/
executor.shutdown();

/*
* Wait for the finalization of the executor
*/
executor.awaitTermination(1, TimeUnit.DAYS);

/*
* Write a message indicating the end of the program
*/
System.out.printf("Main: End of the program.\n");
}

}

package com.packtpub.java7.concurrency.chapter7.recipe05.task;

import java.util.concurrent.TimeUnit;

/**
* Runnable object to check the MyScheduledTask and MyScheduledThreadPoolExecutor classes.
*
*/
public class Task implements Runnable {

/**
* Main method of the task. Writes a message, sleeps the current thread for two seconds and
* writes another message
*/
@Override
public void run() {
System.out.printf("Task: Begin.\n");
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.printf("Task: End.\n");
}

}

package com.packtpub.java7.concurrency.chapter7.recipe05.task;

import java.util.concurrent.RunnableScheduledFuture;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
* Our implementation of an ScheduledThreadPoolExecutor two executes MyScheduledTasks tasks. It extends
* the ScheduledThreadPoolExecutor class
*
*/
public class MyScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor {

/**
* Constructor of the class. Calls the constructor of its parent class using the super keyword
* @param corePoolSize Number of threads to keep in the pool
*/
public MyScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize);
}

/**
* Method that converts a RunnableScheduledFuture task in a MyScheduledTask task
*/
@Override
protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
MyScheduledTask<V> myTask=new MyScheduledTask<V>(runnable, null, task,this);
return myTask;
}

/**
* Method that schedule in the executor a periodic tasks. It calls the method of its parent class using
* the super keyword and stores the period of the task.
*/
@Override
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
ScheduledFuture<?> task= super.scheduleAtFixedRate(command, initialDelay, period, unit);
MyScheduledTask<?> myTask=(MyScheduledTask<?>)task;
myTask.setPeriod(TimeUnit.MILLISECONDS.convert(period,unit));
return task;
}

}

package com.packtpub.java7.concurrency.chapter7.recipe05.task;

import java.util.Date;
import java.util.concurrent.Delayed;
import java.util.concurrent.FutureTask;
import java.util.concurrent.RunnableScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
*
* This class implements an scheduled task to be execute in a scheduled thread pool executor. It's
* a parameterized class where V is the type of data that will be returned by the task.
*
* An scheduled thread pool executor can execute two kinds of tasks:
*      Delayed Tasks: This kind of tasks are executed once after a period of time.
*      Periodic Tasks: This kind of tasks are executed from time to time
* @param <V> Type of data that will be returned by the task
*
*/
public class MyScheduledTask<V> extends FutureTask<V> implements RunnableScheduledFuture<V> {

/**
* Attribute to store the task that will be used to create a MyScheduledTask
*/
private RunnableScheduledFuture<V> task;

/**
* ScheduledThreadPoolExecutor that is going to execute the task
*/
private ScheduledThreadPoolExecutor executor;

/**
* Period of time between two executions of the task
*/
private long period;

/**
* Date when will begin the next execution of the task
*/
private long startDate;

/**
* Constructor of the class. It initializes the attributes of the class
* @param runnable Runnable submitted to be executed by the task
* @param result Result that will be returned by the task
* @param task Task that will execute the Runnable object
* @param executor Executor that is going to execute the task
*/
public MyScheduledTask(Runnable runnable, V result, RunnableScheduledFuture<V> task, ScheduledThreadPoolExecutor executor) {
super(runnable, result);
this.task=task;
this.executor=executor;
}

/**
* Method that returns the reminder for the next execution of the task. If is
* a delayed task, returns the delay of the original task. Else, return the difference
* between the startDate attribute and the actual date.
* @param unit TimeUnit to return the result
*/
@Override
public long getDelay(TimeUnit unit) {
if (!isPeriodic()) {
return task.getDelay(unit);
} else {
if (startDate==0){
return task.getDelay(unit);
} else {
Date now=new Date();
long delay=startDate-now.getTime();
return unit.convert(delay, TimeUnit.MILLISECONDS);
}
}
}

/**
* Method to compare two tasks. It calls the compareTo() method of the original task
*/
@Override
public int compareTo(Delayed o) {
return task.compareTo(o);
}

/**
* Method that returns if the task is periodic or not. It calls the isPeriodic() method
* of the original task
*/
@Override
public boolean isPeriodic() {
return task.isPeriodic();
}

/**
* Method that executes the task. If it's a periodic task, it updates the
* start date of the task and store in the queue of the executor the task to
* be executed again
*/
@Override
public void run() {
if (isPeriodic() && (!executor.isShutdown())) {
Date now=new Date();
startDate=now.getTime()+period;
executor.getQueue().add(this);
}
System.out.printf("Pre-MyScheduledTask: %s\n",new Date());
System.out.printf("MyScheduledTask: Is Periodic: %s\n",isPeriodic());
super.runAndReset();
System.out.printf("Post-MyScheduledTask: %s\n",new Date());
}

/**
* Method that establish the period of the task for periodic tasks
* @param period
*/
public void setPeriod(long period) {
this.period=period;
}
}

结果

有了 Task 类例子总是完成了，它实现 Runnable 接口，也是在计划的执行者中运行的任务。这个例子的主类创建了 MyScheduledThreadPoolExecutor 执行者，然后给他们发送了以下2个任务：

一个延迟任务，在当前时间过一秒后运行

一个周期任务，在当前时间过一秒后运行，接着每隔3秒运行

以下结果展示了这个例子的运行的一部分。你可以检查2种任务运行正常

Main: Mon Jul 25 23:02:05 CST 2016

Pre-MyScheduledTask: Mon Jul 25 23:02:06 CST 2016

MyScheduledTask: Is Periodic: false

Task: Begin.

Main: Mon Jul 25 23:02:08 CST 2016

Task: End.

Post-MyScheduledTask: Mon Jul 25 23:02:08 CST 2016

Pre-MyScheduledTask: Mon Jul 25 23:02:09 CST 2016

MyScheduledTask: Is Periodic: true

Task: Begin.

Task: End.

Post-MyScheduledTask: Mon Jul 25 23:02:11 CST 2016

Pre-MyScheduledTask: Mon Jul 25 23:02:12 CST 2016

MyScheduledTask: Is Periodic: true

Task: Begin.

Task: End.

Post-MyScheduledTask: Mon Jul 25 23:02:14 CST 2016

Pre-MyScheduledTask: Mon Jul 25 23:02:15 CST 2016

MyScheduledTask: Is Periodic: true

Task: Begin.

Task: End.

Post-MyScheduledTask: Mon Jul 25 23:02:17 CST 2016

Main: End of the program.

原理

在这个指南，你实现了 MyScheduledTask 类实现在 ScheduledThreadPoolExecutor 执行者中执行的自定义任务。这个类扩展 FutureTask 类并实现了 RunnableScheduledFuture 接口。它实现 RunnableScheduledFuture 接口， 因为在计划的执行者中执行的全部任务都一定要实现 这个接口，并扩展了 FutureTask 类，因为这个类提供了能有效的实现在 RunnableScheduledFuture 接口声明的方法。 之前提到的全部接口和类都被参数化成任务要返回的数据类型。

为了在计划的执行者中使用 MyScheduledTask 任务，要重写在 MyScheduledThreadPoolExecutor 类的 decorateTask() 方法。这个类扩展 ScheduledThreadPoolExecutor 执行者和它的方法提供一个把 ScheduledThreadPoolExecutor 执行者默认的计划任务转换成 MyScheduledTask 任务来实现的机制。所以，当你实现你的版本的计划任务时，你必须实现你的版本的计划的执行者。

decorateTask() 方法只是简单的创建了新的带有参数的 MyScheduledTask 对象：将要在任务中执行的 Runnable 对象; 将被任务返回结果对象，在这个例子，任务将不会返回结果，所以你要使用null值；原来执行 Runnable 对象的任务，新的对象将在池中代替这个任务；和

将执行任务的执行者，在这个例子，你使用 this 关键词指向创建这个任务的执行者。

The MyScheduledTask 类可以执行延迟和周期性任务。你已经实现了有全部必须的算法可以执行这2种任务的方法。他们是 getDelay() 和 run() 方法。

The getDelay() 方法被计划的执行者调用来确认它是否需要运行任务。此方法对延迟任务和周期任务的响应是不同的。在之前提到的， MyScheduledClass 类的构造函数接收 原先的将要执行 Runnable 对象的 ScheduledRunnableFuture 对象， 并储存它作为类的属性来获取它的方法和它的数据。当我们要运行延迟任务时，getDelay() 方法返回原先任务的延迟，但是在周期任务的例子中，getDelay() 方法返回 startDate 属性值与当前时间的相差值。

run() 方法是用来执行任务的。周期性任务的一个特别之处是你必须把下一次任务的执行作为一个新的任务放入到执行者的queue中，如果你要再次运行任务的话。所以，如果你执行周期性任务，你确定 -

startDate 属性值通过把当前时间和任务的执行周期相加，然后把任务储存在执行者的queue中。startDate 属性储存下一次任务将开始运行的时间。然后，使用 FutureTask 类提供的 runAndReset() 方法来运行任务。 在这个例子的延迟任务由于他们仅仅执行一次，就不用把他们放入执行者的queue中了。

你必须要注意如果执行者已经关闭。在这个例子，你不不需要再次把周期性任务储存进执行者的queue。

最后，你重写了在 MyScheduledThreadPoolExecutor 类的 scheduleAtFixedRate() 方法。我们之前提到的，对于周期任务，你要使用任务的周期来确定 startDate 属性值，但是你还没有初始这个周期呢。你必须重写此方法接收周期作为参数，然后传递给 MyScheduledTask 类这样它才能使用。

扩展

ScheduledThreadPoolExecutor 类提供了另一个版本的 decorateTask() 方法，它接收 Callable 对象作为参数来代替 Runnable 对象。

转自：ifeve