定制并发类（六）自定义在计划的线程池内运行的任务

2021-10-21 20:30:43

声明：本文是《 Java 7 Concurrency Cookbook 》的第七章， 作者： Javier Fernández González 译者：郑玉婷

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

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

  • Delayed 任务：这种任务在一段时间后仅执行一次。
  • Periodic 任务：这种任务在延迟后执行，然后通常周期性运行

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

准备

指南中的例子是使用Eclipse IDE 来实现的。如果你使用Eclipse 或者其他的IDE，例如NetBeans, 打开并创建一个新的java项目。

怎么做呢…

按照这些步骤来实现下面的例子：


001 //1.  创建一个类，名为 MyScheduledTask，使名为 V 的泛型类型参数化。它扩展 FutureTask 类并实现 RunnableScheduledFuture 接口。
002 public class MyScheduledTask<V> extends FutureTask<V> implements
003 RunnableScheduledFuture<V> {
004  
005 //2.   声明一个私有 RunnableScheduledFuture 属性，名为 task.
006 private RunnableScheduledFuture<V> task;
007  
008 //3.   声明一个私有 ScheduledThreadPoolExecutor，名为 executor.
009 private ScheduledThreadPoolExecutor executor;
010  
011 //4.   声明一个私有long属性，名为 period。
012 private long period;
013  
014 //5.   声明一个私有long属性，名为 startDate。
015 private long startDate;
016  
017 //6.   实现类的构造函数。它接收任务：将要运行的 Runnable 对象，任务要返回的 result，将被用来创建 MyScheduledTask 对象的 RunnableScheduledFuture 任务，和要执行这个任务的 ScheduledThreadPoolExecutor 对象。 调用它的父类的构造函数并储存任务和执行者属性。
018 public MyScheduledTask(Runnable runnable, V result, RunnableScheduledFuture<V> task, ScheduledThreadPoolExecutor executor) {
019     super(runnable, result);
020     this.task=task;
021     this.executor=executor;
022 }
023  
024 //7.    实现 getDelay() 方法。如果是周期性任务且 startDate 形象的值非0，计算并返回 startDate 属性与当前日期的相差值。否则，返回储存在 task 属性的原先任务的延迟值。不要忘记你要返回结果时，要传递 time unit 作为参数哦。
025 @Override
026 public long getDelay(TimeUnit unit) {
027     if (!isPeriodic()) {
028         return task.getDelay(unit);
029     } else {
030         if (startDate==0){
031             return task.getDelay(unit);
032         } else {
033             Date now=new Date();
034             long delay=startDate-now.getTime();
035             return unit.convert(delay, TimeUnit.MILLISECONDS);
036         }
037     }
038 }
039  
040 //8.  实现 compareTo() 方法。调用原先任务的 compareTo() 方法。
041 @Override
042 public int compareTo(Delayed o) {
043     return task.compareTo(o);
044 }
045  
046 //9.  实现 isPeriodic() 方法。调用原来任务的 isPeriodic() 方法。
047 @Override
048 public boolean isPeriodic() {
049 return task.isPeriodic();
050 }
051  
052 //10. 实现方法 run()。如果这是一个周期性任务，你要用下一个执行任务的开始日期更新它的 startDate 属性。用当前日期和时间间隔的和计算它。 然后，把再次把任务添加到 ScheduledThreadPoolExecutor 对象的 queue中。
053 @Override
054 public void run() {
055     if (isPeriodic() && (!executor.isShutdown())) {
056         Date now=new Date();
057         startDate=now.getTime()+period;
058         executor.getQueue().add(this);
059     }
060  
061 //11.打印当前日期的信息到操控台，调用 runAndReset() 方法运行任务，然后再打印另一条关于当前日期的信息到操控台。
062 System.out.printf("Pre-MyScheduledTask: %s\n",new Date());
063 System.out.printf("MyScheduledTask: Is Periodic:%s\n",isPeriodic());
064 super.runAndReset();
065 System.out.printf("Post-MyScheduledTask: %s\n",new Date());
066 }
067  
068 //12. 实现 setPeriod() 方法，来确立任务的周期时间。
069 public void setPeriod(long period) {
070     this.period=period;
071 }
072  
073 //13. 创建一个类，名为 MyScheduledThreadPoolExecutor 来实现一个运行 MyScheduledTask 任务的 ScheduledThreadPoolExecutor 对象。特别扩展 ScheduledThreadPoolExecutor 类。
074 public class MyScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor {
075  
076 //14. 实现类的构造函数，只要调用它的父类的构造函数。
077 public MyScheduledThreadPoolExecutor(int corePoolSize) {
078     super(corePoolSize);
079 }
080  
081 //15. 实现方法 decorateTask()。它接收将要被运行的 Runnable 对象和将运行 Runnable 对象的 RunnableScheduledFuture 任务作为参数。使用这些对象来构造来创建并返回 MyScheduledTask 任务。
082 @Override
083 //译者：前面那个<V>是打错吧多余的吧？
084 protected <V> RunnableScheduledFuture<V> decorateTask(Runnable runnable, RunnableScheduledFuture<V> task) {
085     MyScheduledTask<V> myTask=new MyScheduledTask<V>(runnable, null, task,this);
086     return myTask;
087 }
088  
089 //16. 覆盖方法 scheduledAtFixedRate()。调用它的父类的方法，调用它的父类的方法，  method. Call the method of its parent class, convert the returned object into a MyScheduledTask object, and establish the period of that task using the setPeriod() method.
090  
091 @Override
092 //译者：不知道怎么出现？号的。应该是V。
093 public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
094     ScheduledFuture<?> task= super.scheduleAtFixedRate(command, initialDelay, period, unit);
095     MyScheduledTask<?> myTask=(MyScheduledTask<?>)task;
096     myTask.setPeriod(TimeUnit.MILLISECONDS.convert(period,unit));
097     return task;
098 }
099  
100 //17.  创建一个类，名为 Task，实现 Runnable 接口。
101 public class Task implements Runnable {
102  
103 //18. 实现方法 run() 。在任务开始时打印一条信息，再让当前线程进入休眠2秒。最后在任务结束时，再打印另一条信息。
104 @Override
105 public void run() {
106     System.out.printf("Task: Begin.\n");
107     try {
108         TimeUnit.SECONDS.sleep(2);
109     } catch (InterruptedException e) {
110         e.printStackTrace();
111     }
112     System.out.printf("Task: End.\n");
113 }
114  
115 //19. 创建例子的主类通过创建一个类，名为 Main 并添加 main()方法。
116 public class Main {
117  
118 public static void main(String[] args) throws Exception{
119  
120 //20. 创建一个 MyScheduledThreadPoolExecutor 对象，名为 executor。使用2作为参数来在池中获得2个线程。
121 MyScheduledThreadPoolExecutor executor=new MyScheduledThreadPoolExecutor(2);
122  
123 //21. 创建 Task 对象，名为 task。把当前日期写入操控台。
124 Task task=new Task(); System.out.printf("Main: %s\n",new Date());
125  
126 //22. 使用 schedule() 方法发送一个延迟任务给执行者。此任务在延迟一秒后运行。
127 executor.schedule(task, 1, TimeUnit.SECONDS);
128  
129 //23. 让主线程休眠3秒。
130 TimeUnit.SECONDS.sleep(3);
131  
132 //24. 创建另一个 Task 对象。再次在操控台打印当前日期。
133 task=new Task();
134 System.out.printf("Main: %s\n",new Date());
135  
136 //25. 使用方法 scheduleAtFixedRate()发送一个周期性任务给执行者。此任务在延迟一秒后被运行，然后每3秒执行。
137 executor.scheduleAtFixedRate(task, 1, 3, TimeUnit.SECONDS);
138  
139 //26. 让主线程休眠10秒。
140 TimeUnit.SECONDS.sleep(10);
141  
142 //27. 使用 shutdown() 方法关闭执行者。使用 awaitTermination() 方法等待执行者的完结。
143 executor.shutdown();
144 executor.awaitTermination(1, TimeUnit.DAYS);
145  
146 //28. 写信息到操控台表明任务结束。
147 System.out.printf("Main: End of the program.\n");

它是怎么工作的…

在这个指南，你实现了 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 类这样它才能使用。

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

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

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

定制并发类（六）自定义在计划的线程池内运行的任务

更多…

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

上一篇：基于框架式的短视频APP开发 SDK 架构设计实践


下一篇：Python也有保质期？它在脚本领域的“大佬地位”还能持续多久？