【发布时间】:2014-08-19 20:39:27
【问题描述】:
我在使用ExecutorCompletionService 时遇到了以下问题。我想在不同的线程中调用很多 Callable。这些 Callable 之间不共享任何信息。我需要为每个 Callable 定义一个超时,例如。运行时间不要超过 5 秒。每个 Callable 可以在我不知道何时开始的不同时间运行。超时后,应该停止/杀死线程,结果对我来说不再有趣。不应影响其他“正常”运行的线程。
因此,让我们以一个简单的可调用对象和我当前的 java 代码为例。
import java.util.Date;
import java.util.concurrent.Callable;
public class Job implements Callable<Integer> {
int returnValue = 0;
long millis = 0;
public Job(long millis, int value) {
this.millis = millis;
this.returnValue = value;
}
@Override
public Integer call() throws Exception, InterruptedException {
try {
System.out.println(new Date() + " " + returnValue + " started");
Thread.sleep(millis);
System.out.println(new Date() + " " + returnValue + " finished");
return returnValue;
} catch (InterruptedException e) {
System.out.println(new Date() + " " + returnValue + " interrupted");
throw e;
}
}
}
以及使用 Callable 的另一个类。
import java.util.ArrayList;
import java.util.Date;
import java.util.concurrent.*;
public class CallableTest {
public static void main(String[] args) {
ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(2);
CompletionService<Integer> pool = new ExecutorCompletionService<Integer>(newFixedThreadPool);
for (int i = 10; i > 0; i--) {
Job job = new Job(i * 1000, i);
pool.submit(job);
}
ArrayList<Integer> results = new ArrayList<Integer>();
for (int i = 1; i < 11; ++i) {
try {
Future<Integer> future = pool.take();
Integer content = future.get(5, TimeUnit.SECONDS);
results.add(content);
System.out.println(new Date() + " added " + content);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
} catch (Exception e) {
e.printStackTrace();
}
}
newFixedThreadPool.shutdownNow();
System.out.println(new Date() + " results:");
for (int j : results) {
System.out.println(new Date() + " " + j);
}
}
}
输出类似于:
Sun Jun 29 08:01:00 CEST 2014 10 started
Sun Jun 29 08:01:00 CEST 2014 9 started
Sun Jun 29 08:01:09 CEST 2014 9 finished
Sun Jun 29 08:01:09 CEST 2014 added 9
Sun Jun 29 08:01:09 CEST 2014 8 started
Sun Jun 29 08:01:10 CEST 2014 10 finished
Sun Jun 29 08:01:10 CEST 2014 7 started
Sun Jun 29 08:01:10 CEST 2014 added 10
Sun Jun 29 08:01:17 CEST 2014 7 finished
Sun Jun 29 08:01:17 CEST 2014 6 started
Sun Jun 29 08:01:17 CEST 2014 added 7
Sun Jun 29 08:01:17 CEST 2014 8 finished
Sun Jun 29 08:01:17 CEST 2014 added 8
Sun Jun 29 08:01:17 CEST 2014 5 started
Sun Jun 29 08:01:22 CEST 2014 5 finished
Sun Jun 29 08:01:22 CEST 2014 added 5
Sun Jun 29 08:01:22 CEST 2014 4 started
Sun Jun 29 08:01:23 CEST 2014 6 finished
Sun Jun 29 08:01:23 CEST 2014 3 started
Sun Jun 29 08:01:23 CEST 2014 added 6
Sun Jun 29 08:01:26 CEST 2014 3 finished
Sun Jun 29 08:01:26 CEST 2014 2 started
Sun Jun 29 08:01:26 CEST 2014 added 3
Sun Jun 29 08:01:26 CEST 2014 4 finished
Sun Jun 29 08:01:26 CEST 2014 1 started
Sun Jun 29 08:01:26 CEST 2014 added 4
Sun Jun 29 08:01:27 CEST 2014 1 finished
Sun Jun 29 08:01:27 CEST 2014 added 1
Sun Jun 29 08:01:28 CEST 2014 2 finished
Sun Jun 29 08:01:28 CEST 2014 added 2
Sun Jun 29 08:01:28 CEST 2014 results:
Sun Jun 29 08:01:28 CEST 2014 9
Sun Jun 29 08:01:28 CEST 2014 10
Sun Jun 29 08:01:28 CEST 2014 7
Sun Jun 29 08:01:28 CEST 2014 8
Sun Jun 29 08:01:28 CEST 2014 5
Sun Jun 29 08:01:28 CEST 2014 6
Sun Jun 29 08:01:28 CEST 2014 3
Sun Jun 29 08:01:28 CEST 2014 4
Sun Jun 29 08:01:28 CEST 2014 1
Sun Jun 29 08:01:28 CEST 2014 2
这不像我想要的那样工作。我希望每个运行时间超过 5 秒的 Callable 都应该被终止/结束/中断,并且只有运行时间低于 5 秒的 Callable 才能给我一个有效的结果。
我也试过没有 ExecutorCompletionService
public class CallableTest2 {
public static void main(String[] args) {
ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(2);
List<Future<Integer>> futures = new ArrayList<Future<Integer>>();
for (int i = 10; i > 0; i--) {
Job job = new Job(i * 1000, i);
futures.add(newFixedThreadPool.submit(job));
}
ArrayList<Integer> results = new ArrayList<Integer>();
for (Future<Integer> future: futures) {
try {
Integer content = future.get(5, TimeUnit.SECONDS);
results.add(content);
System.out.println(new Date() + " added " + content);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
} catch (Exception e) {
e.printStackTrace();
}
}
newFixedThreadPool.shutdownNow();
System.out.println(new Date() + " results:");
for (int j : results) {
System.out.println(new Date() + " " + j);
}
}
}
有了结果:
Sun Jun 29 08:33:19 CEST 2014 9 started
Sun Jun 29 08:33:19 CEST 2014 10 started
java.util.concurrent.TimeoutException
at java.util.concurrent.FutureTask$Sync.innerGet(FutureTask.java:228)
at java.util.concurrent.FutureTask.get(FutureTask.java:91)
at callabletest.CallableTest2.main(CallableTest2.java:29)
Sun Jun 29 08:33:28 CEST 2014 9 finished
Sun Jun 29 08:33:28 CEST 2014 8 started
Sun Jun 29 08:33:28 CEST 2014 added 9
Sun Jun 29 08:33:29 CEST 2014 10 finished
Sun Jun 29 08:33:29 CEST 2014 7 started
java.util.concurrent.TimeoutException
at java.util.concurrent.FutureTask$Sync.innerGet(FutureTask.java:228)
at java.util.concurrent.FutureTask.get(FutureTask.java:91)
at callabletest.CallableTest2.main(CallableTest2.java:29)
Sun Jun 29 08:33:36 CEST 2014 7 finished
Sun Jun 29 08:33:36 CEST 2014 added 7
Sun Jun 29 08:33:36 CEST 2014 6 started
Sun Jun 29 08:33:36 CEST 2014 8 finished
Sun Jun 29 08:33:36 CEST 2014 5 started
java.util.concurrent.TimeoutException
at java.util.concurrent.FutureTask$Sync.innerGet(Sun Jun 29 08:33:41 CEST 2014 5 finished
FutureTask.java:228)
Sun Jun 29 08:33:41 CEST 2014 added 5
at java.util.concurrent.FutureTask.get(FutureTask.java:91)
Sun Jun 29 08:33:41 CEST 2014 4 started
at callabletest.CallableTest2.main(CallableTest2.java:29)
Sun Jun 29 08:33:42 CEST 2014 6 finished
Sun Jun 29 08:33:42 CEST 2014 3 started
Sun Jun 29 08:33:45 CEST 2014 3 finished
Sun Jun 29 08:33:45 CEST 2014 2 started
Sun Jun 29 08:33:45 CEST 2014 4 finished
Sun Jun 29 08:33:45 CEST 2014 added 4
Sun Jun 29 08:33:45 CEST 2014 added 3
Sun Jun 29 08:33:45 CEST 2014 1 started
Sun Jun 29 08:33:46 CEST 2014 1 finished
Sun Jun 29 08:33:47 CEST 2014 2 finished
Sun Jun 29 08:33:47 CEST 2014 added 2
Sun Jun 29 08:33:47 CEST 2014 added 1
Sun Jun 29 08:33:47 CEST 2014 results:
Sun Jun 29 08:33:47 CEST 2014 9
Sun Jun 29 08:33:47 CEST 2014 7
Sun Jun 29 08:33:47 CEST 2014 5
Sun Jun 29 08:33:47 CEST 2014 4
Sun Jun 29 08:33:47 CEST 2014 3
Sun Jun 29 08:33:47 CEST 2014 2
Sun Jun 29 08:33:47 CEST 2014 1
现在我得到了一些 TimeoutExceptions,但也不是我期望的。例如。运行 9 和 7 秒的 Callable 不会抛出异常!
我必须在代码中进行哪些更改,以仅获取短运行线程的结果并终止长时间运行的线程。在示例中只有结果 1-5 而没有 6-10。
我已经测试了很多东西,但我无法让它发挥作用。 请帮忙
这是对使用 ScheduledExecutorService 的 bstar55 帖子的回答。
我将关于您的提示的代码更改为:
public class CallableTest3 {
public static void main(String[] args) {
ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);
List<Future<Integer>> futures = new ArrayList<Future<Integer>>();
for (int i = 10; i > 0; i--) {
Job job = new Job(i * 1000, i);
final Future handler = executor.submit(job);
final int x = i;
executor.schedule(new Runnable() {
public void run() {
boolean cancel = handler.cancel(true);
if(cancel){
System.out.println(new Date() + " job " + x + " cancelled");
}else{
System.out.println(new Date() + " job " + x + " not cancelled");
}
}
}, 5000, TimeUnit.MILLISECONDS);
futures.add(handler);
}
ArrayList<Integer> results = new ArrayList<Integer>();
for (Future<Integer> future : futures) {
try {
Integer content = future.get(5, TimeUnit.SECONDS);
results.add(content);
System.out.println(new Date() + " added " + content);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
} catch (Exception e) {
e.printStackTrace();
}
}
executor.shutdown();
System.out.println(new Date() + " results:");
for (int j : results) {
System.out.println(new Date() + " --- " + j);
}
}
}
但这也不能按预期工作。 结果:
Sun Jun 29 10:27:41 CEST 2014 9 started
Sun Jun 29 10:27:41 CEST 2014 10 started
java.util.concurrent.TimeoutException
at java.util.concurrent.FutureTask$Sync.innerGet(FutureTask.java:228)
at java.util.concurrent.FutureTask.get(FutureTask.java:91)
at callabletest.CallableTest3.main(CallableTest3.java:43)
Sun Jun 29 10:27:50 CEST 2014 9 finished
Sun Jun 29 10:27:50 CEST 2014 added 9
Sun Jun 29 10:27:50 CEST 2014 8 started
Sun Jun 29 10:27:51 CEST 2014 10 finished
Sun Jun 29 10:27:51 CEST 2014 7 started
java.util.concurrent.TimeoutException
at java.util.concurrent.FutureTask$Sync.innerGet(FutureTask.java:228)
at java.util.concurrent.FutureTask.get(FutureTask.java:91)
at callabletest.CallableTest3.main(CallableTest3.java:43)
Sun Jun 29 10:27:58 CEST 2014 8 finished
Sun Jun 29 10:27:58 CEST 2014 6 started
Sun Jun 29 10:27:58 CEST 2014 7 finished
Sun Jun 29 10:27:58 CEST 2014 5 started
Sun Jun 29 10:27:58 CEST 2014 added 7
Sun Jun 29 10:28:03 CEST 2014 5 finished
Sun Jun 29 10:28:03 CEST 2014 4 started
java.util.concurrent.TimeoutException
at java.util.concurrent.FutureTask$Sync.innerGet(FutureTask.java:228)
at java.util.concurrent.FutureTask.get(FutureTask.java:91)
Sun Jun 29 10:28:03 CEST 2014 added 5
at callabletest.CallableTest3.main(CallableTest3.java:43)
Sun Jun 29 10:28:04 CEST 2014 6 finished
Sun Jun 29 10:28:04 CEST 2014 3 started
Sun Jun 29 10:28:07 CEST 2014 3 finished
Sun Jun 29 10:28:07 CEST 2014 2 started
Sun Jun 29 10:28:07 CEST 2014 4 finished
Sun Jun 29 10:28:07 CEST 2014 added 4
Sun Jun 29 10:28:07 CEST 2014 added 3
Sun Jun 29 10:28:07 CEST 2014 1 started
java.util.concurrent.CancellationException
at java.util.concurrent.FutureTask$Sync.innerGet(FutureTask.java:230)
at java.util.concurrent.FutureTask.get(FutureTask.java:91)
at callabletest.CallableTest3.main(CallableTest3.java:43)
Sun Jun 29 10:28:08 CEST 2014 1 finished
Sun Jun 29 10:28:08 CEST 2014 job 10 not cancelled
Sun Jun 29 10:28:08 CEST 2014 job 9 not cancelled
Sun Jun 29 10:28:08 CEST 2014 job 8 not cancelled
Sun Jun 29 10:28:08 CEST 2014 job 7 not cancelled
Sun Jun 29 10:28:08 CEST 2014 job 6 not cancelled
Sun Jun 29 10:28:08 CEST 2014 job 5 not cancelled
Sun Jun 29 10:28:08 CEST 2014 job 4 not cancelled
Sun Jun 29 10:28:08 CEST 2014 job 3 not cancelled
Sun Jun 29 10:28:08 CEST 2014 2 interrupted
Sun Jun 29 10:28:08 CEST 2014 job 1 not cancelled
Sun Jun 29 10:28:08 CEST 2014 added 1
Sun Jun 29 10:28:08 CEST 2014 results:
Sun Jun 29 10:28:08 CEST 2014 --- 9
Sun Jun 29 10:28:08 CEST 2014 --- 7
Sun Jun 29 10:28:08 CEST 2014 --- 5
Sun Jun 29 10:28:08 CEST 2014 --- 4
Sun Jun 29 10:28:08 CEST 2014 --- 3
Sun Jun 29 10:28:08 CEST 2014 --- 1
Sun Jun 29 10:28:08 CEST 2014 job 2 cancelled
但是工作 2 被取消了!
【问题讨论】:
-
记录何时提交作业以及何时从池中获取作业。如果有兴趣查看在调用 future.get() 之前第一个作业已经运行了多长时间。
-
好的,我想我开始明白发生了什么。您在两个线程上运行作业,然后尝试在单个线程上获得结果。在您等待一个结果的同时,至少还有一个其他作业正在运行,而您没有等待它。当您开始等待结果时,您等待的作业已经运行了一段时间,使您的 5 秒超时无效。
标签: java multithreading timeout executorservice callable