use*_*186 7 java multithreading timeout callable executorservice
我使用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;
}
}
}
Run Code Online (Sandbox Code Playgroud)
以及使用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);
}
}
}
Run Code Online (Sandbox Code Playgroud)
输出是这样的:
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
Run Code Online (Sandbox Code Playgroud)
这不像我想要的那样有效.我希望每个Callable运行时间超过5秒应该终止/结束/ interruped只有Callable运行低于5秒才能给我一个有效的结果.
我也尝试过没有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);
}
}
}
Run Code Online (Sandbox Code Playgroud)
结果如下:
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
Run Code Online (Sandbox Code Playgroud)
现在我得到一些TimeoutExceptions,但也不是我期望的那些.例如.Callable运行9和7秒不会抛出异常!
我需要在代码中更改什么,才能获得短期运行的线程的结果并杀死长时间运行的线程.在示例中仅结果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);
}
}
}
Run Code Online (Sandbox Code Playgroud)
但这也不能按预期工作.结果:
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
Run Code Online (Sandbox Code Playgroud)
但相反,工作2被取消了!
小智 7
我建议您将问题分为两个独立的问题:
对于第一个(多线程),您已经使用了可以在 2 个线程上管理它的服务执行器:Executors.newFixedThreadPool(2)。如果您在此处应用超时,则超时适用于所有任务的运行,但每个作业都需要超时。
对于超时问题,您可以通过类中每个作业的新服务执行器来管理它:JobManager。
package com.stackoverflow.q24473796;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
public class JobManager implements Callable<Integer> {
protected long timeout;
protected TimeUnit timeUnit;
protected Callable<Integer> job;
public JobManager(long timeout, TimeUnit timeUnit, Callable<Integer> job) {
this.timeout = timeout;
this.timeUnit = timeUnit;
this.job = job;
}
@Override
public Integer call() {
Integer result = new Integer(-1); // default, this could be adapted
ExecutorService exec = Executors.newSingleThreadExecutor();
try {
result = exec.submit(job).get(timeout, timeUnit);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
// Whatever you want
if (e instanceof TimeoutException) {
System.out.println("Timeout get for " + job.toString());
} else {
System.out.println("exception get for " + job.toString() + " : " + e.getMessage());
}
}
exec.shutdown();
return result;
}
}
Run Code Online (Sandbox Code Playgroud)
然后,您可以从主线程调用任务,如下所示:
Job job = new Job(i * 1000, i);
Future<Integer> future = newFixedThreadPool.submit(new JobManager(5, TimeUnit.SECONDS, job));
Run Code Online (Sandbox Code Playgroud)
我调整了您的 CallableTest:包 com.stackoverflow.q24473796;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
public class CallableTest {
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);
Future<Integer> future = newFixedThreadPool.submit(new JobManager(5, TimeUnit.SECONDS, job));
futures.add(future);
}
ArrayList<Integer> results = new ArrayList<Integer>();
for (Future<Integer> future : futures) {
Integer result = new Integer(-1);
try {
result = future.get();
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
if (result != -1) {
results.add(result);
}
}
newFixedThreadPool.shutdown();
try {
newFixedThreadPool.awaitTermination(60, TimeUnit.SECONDS); //Global Timeout
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(new Date() + " results:");
for (int j : results) {
System.out.println(new Date() + " " + j);
}
}
}
Run Code Online (Sandbox Code Playgroud)
您将得到以下输出:
Wed Apr 29 10:51:02 CEST 2015 10 started
Wed Apr 29 10:51:02 CEST 2015 9 started
Timeout get for com.stackoverflow.q24473796.Job@249fe45c
Timeout get for com.stackoverflow.q24473796.Job@249fe45c
Wed Apr 29 10:51:07 CEST 2015 8 started
Wed Apr 29 10:51:07 CEST 2015 7 started
Wed Apr 29 10:51:11 CEST 2015 9 finished
Timeout get for com.stackoverflow.q24473796.Job@3cd4c5a0
Timeout get for com.stackoverflow.q24473796.Job@3cd4c5a0
Wed Apr 29 10:51:12 CEST 2015 6 started
Wed Apr 29 10:51:12 CEST 2015 5 started
Wed Apr 29 10:51:12 CEST 2015 10 finished
Wed Apr 29 10:51:14 CEST 2015 7 finished
Wed Apr 29 10:51:15 CEST 2015 8 finished
Wed Apr 29 10:51:17 CEST 2015 5 finished
Wed Apr 29 10:51:17 CEST 2015 4 started
Timeout get for com.stackoverflow.q24473796.Job@2a0fded2
Wed Apr 29 10:51:17 CEST 2015 3 started
Wed Apr 29 10:51:18 CEST 2015 6 finished
Wed Apr 29 10:51:20 CEST 2015 3 finished
Wed Apr 29 10:51:20 CEST 2015 2 started
Wed Apr 29 10:51:21 CEST 2015 4 finished
Wed Apr 29 10:51:21 CEST 2015 1 started
Wed Apr 29 10:51:22 CEST 2015 1 finished
Wed Apr 29 10:51:22 CEST 2015 2 finished
Wed Apr 29 10:51:22 CEST 2015 results:
Wed Apr 29 10:51:22 CEST 2015 5
Wed Apr 29 10:51:22 CEST 2015 4
Wed Apr 29 10:51:22 CEST 2015 3
Wed Apr 29 10:51:22 CEST 2015 2
Wed Apr 29 10:51:22 CEST 2015 1
Run Code Online (Sandbox Code Playgroud)
也许最好的方法是在 Callable 的 call() 方法中处理计时。当您输入方法时,记录时间。在 call() 方法中分割时间 - 在示例情况下,不是完全休眠,而是一次休眠一秒 - 并且每次都检查自进入时间以来是否已经过了最长时间。如果有,则中止处理并正常退出。
如果您绝对无法按时定期检查 Callable 的代码 - 或者像其他人建议的那样根据外部停止通知进行检查 - 有一种方法可以从线程外部执行的代码终止它。然而,这种方式已被弃用,因为它几乎肯定会导致无法修复的错误。本质上,当线程启动时 - 即当进入 call() 方法时 - 启动一个 TimerTask 子类并给它一个指向线程的指针。当 TimerTask 触发时,它会在线程上调用 Thread.stop()。同样,这是非常不安全的,因为它可能会使对象处于损坏状态,如 Oracle 文档中所述:
http://docs.oracle.com/javase/8/docs/technotes/guides/concurrency/threadPrimitiveDeprecation.html
唯一可行的方法是,如果您的 Callable 的代码永远不需要同步,并且永远不会在任何对象上实际同步,即使在库调用中也是如此,并且将来永远不会更改这样做。这是非常难以确保的,因此最好找到一种方法来分解 Callable 的代码,以便它可以定期检查时间并在超时时优雅地退出。
最后一点:如果您确实认为 Callable 的代码永远不需要同步来共享内存中的对象,那么它是在外部进程中执行的一个很好的候选者,而不是在主进程中的线程中。这样做的优点是您可以相对安全地终止外部进程。基本上,外部进程中的 TimerTask 会调用工作进程中的 System.exit(),而不是工作线程中的 Thread.stop()。
| 归档时间: |
|
| 查看次数: |
6400 次 |
| 最近记录: |