Pau*_*lor 5 java timeout future executorservice
我提出一些使用裹着一个2线FixedThreadPool的ExecutorService一个CompletionService未来的某个任务,我设置,则设置等于提交的任务数的循环和使用completionservice.take()等待他们全部完成或失败.麻烦很偶然,它永远不会完成(但我不知道为什么)所以我将take()方法更改为轮询(300,Timeout.SECONDS),这个想法是,如果一个任务需要超过5分钟才能完成轮询将失败,然后最终将退出循环,我可以通过所有的期货并调用future.cancel(true)强制取消违规任务.
但是当我运行代码并且它挂起时,我看到轮询每5分钟就会连续失败并且没有更多任务运行所以我认为这两个工作人员在某种程度上陷入僵局并且从未完成,并且从不允许其他任务启动.因为超时是5分钟并且仍然有1000个任务要运行所以打破循环所花费的时间太长所以取消了这个工作.
所以我想做的是中断/强制取消当前任务,如果还没有在5分钟内完成,但我无法看到任何方法.
此代码示例显示了我正在谈论的内容的简化版本
import com.jthink.jaikoz.exception.JaikozException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;
public class CompletionServiceTest
{
public static void main(final String[] args)
{
CompletionService<Boolean> cs = new ExecutorCompletionService<Boolean>(Executors.newFixedThreadPool(2));
Collection<Worker> tasks = new ArrayList<Worker>(10);
tasks.add(new Worker(1));
tasks.add(new Worker(2));
tasks.add(new Worker(3));
tasks.add(new Worker(4));
tasks.add(new Worker(5));
tasks.add(new Worker(6));
List<Future<Boolean>> futures = new ArrayList<Future<Boolean>>(tasks.size());
try
{
for (Callable task : tasks)
{
futures.add(cs.submit(task));
}
for (int t = 0; t < futures.size(); t++)
{
Future<Boolean> result = cs.poll(10, TimeUnit.SECONDS);
if(result==null)
{
System.out.println("Worker TimedOut:");
continue;
}
else
{
try
{
if(result.isDone() && result.get())
{
System.out.println("Worker Completed:");
}
else
{
System.out.println("Worker Failed");
}
}
catch (ExecutionException ee)
{
ee.printStackTrace();
}
}
}
}
catch (InterruptedException ie)
{
}
finally
{
//Cancel by interrupting any existing tasks currently running in Executor Service
for (Future<Boolean> f : futures)
{
f.cancel(true);
}
}
System.out.println("Done");
}
}
class Worker implements Callable<Boolean>
{
private int number;
public Worker(int number)
{
this.number=number;
}
public Boolean call()
{
if(number==3)
{
try
{
Thread.sleep(50000);
}
catch(InterruptedException tie)
{
}
}
return true;
}
}
产量
Worker Completed:
Worker Completed:
Worker Completed:
Worker Completed:
Worker Completed:
Worker TimedOut:
Done
我想我已经解决了它，基本上如果发生超时，我然后遍历我的未来对象列表并找到第一个尚未完成的对象，并强制取消。看起来不那么优雅，但似乎有效。
我改变了池的大小只是为了显示更好地演示解决方案的输出，但也适用于 2 个线程池。
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;
public class CompletionServiceTest
{
public static void main(final String[] args)
{
CompletionService<Boolean> cs = new ExecutorCompletionService<Boolean>(Executors.newFixedThreadPool(1));
Collection<Worker> tasks = new ArrayList<Worker>(10);
tasks.add(new Worker(1));
tasks.add(new Worker(2));
tasks.add(new Worker(3));
tasks.add(new Worker(4));
tasks.add(new Worker(5));
tasks.add(new Worker(6));
List<Future<Boolean>> futures = new ArrayList<Future<Boolean>>(tasks.size());
try
{
for (Callable task : tasks)
{
futures.add(cs.submit(task));
}
for (int t = 0; t < futures.size(); t++)
{
System.out.println("Invocation:"+t);
Future<Boolean> result = cs.poll(10, TimeUnit.SECONDS);
if(result==null)
{
System.out.println(new Date()+":Worker Timedout:");
//So lets cancel the first futures we find that havent completed
for(Future future:futures)
{
System.out.println("Checking future");
if(future.isDone())
{
continue;
}
else
{
future.cancel(true);
System.out.println("Cancelled");
break;
}
}
continue;
}
else
{
try
{
if(result.isDone() && !result.isCancelled() && result.get())
{
System.out.println(new Date()+":Worker Completed:");
}
else if(result.isDone() && !result.isCancelled() && !result.get())
{
System.out.println(new Date()+":Worker Failed");
}
}
catch (ExecutionException ee)
{
ee.printStackTrace(System.out);
}
}
}
}
catch (InterruptedException ie)
{
}
finally
{
//Cancel by interrupting any existing tasks currently running in Executor Service
for (Future<Boolean> f : futures)
{
f.cancel(true);
}
}
System.out.println(new Date()+":Done");
}
}
class Worker implements Callable<Boolean>
{
private int number;
public Worker(int number)
{
this.number=number;
}
public Boolean call()
throws InterruptedException
{
try
{
if(number==3)
{
Thread.sleep(50000);
}
}
catch(InterruptedException ie)
{
System.out.println("Worker Interuppted");
throw ie;
}
return true;
}
}
输出是
Invocation:0
Thu Mar 10 20:51:39 GMT 2011:Worker Completed:
Invocation:1
Thu Mar 10 20:51:39 GMT 2011:Worker Completed:
Invocation:2
Thu Mar 10 20:51:49 GMT 2011:Worker Timedout:
Checking future
Checking future
Checking future
Cancelled
Invocation:3
Worker Interuppted
Invocation:4
Thu Mar 10 20:51:49 GMT 2011:Worker Completed:
Invocation:5
Thu Mar 10 20:51:49 GMT 2011:Worker Completed:
Thu Mar 10 20:51:49 GMT 2011:Done