java ExecutorService如何处理超时

Question

我正在尝试创建一个用于同时调用多个 Web 服务的存根，但在处理 CancellationException 时出现错误。这里是主要方法

    ExecutorService pool= Executors.newFixedThreadPool(7);
    List<Future<Long>> futureList = new ArrayList<Future<Long>>();
    Set<CallableDemo> callList = new HashSet<CallableDemo>();

    callList.add(new CallableDemo(0L));
    callList.add(new CallableDemo(10L));
    callList.add(new CallableDemo(20L));
    callList.add(new CallableDemo(30L));
    callList.add(new CallableDemo(40L));
    callList.add(new CallableDemo(50L));
    callList.add(new CallableDemo(-600L));
    callList.add(new CallableDemo(-700L));
    callList.add(new CallableDemo(-800L));
    callList.add(new CallableDemo(-900L));

    futureList = pool.invokeAll(callList, 15L, TimeUnit.SECONDS);

    for(Future<Long> fut : futureList){
    try {
            System.out.println(new Date()+ "::"+fut.get());
    } catch (InterruptedException e) {
         System.out.println("Done :)");
         e.printStackTrace();
        Thread.currentThread().interrupt();
    }
    catch (ExecutionException e) {
         System.out.println("Done :)");
        e.printStackTrace();
        Thread.currentThread().interrupt();
    }
    }
    executor.shutdown();

这里是 CallableDemo，

 import java.util.concurrent.Callable;
 public class CallableDemo implements Callable<Long>
{
private Long count = 0L;

public CallableDemo(Long i)
{
    this.count = i;
}

public Long call() throws Exception
{
    Long i;
    for( i = this.count; i < 100L; i++)
    {
        try { Thread.sleep(100); }
            catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return i;
        }
        System.out.println(Thread.currentThread().getName() + " - " + i);
    }
    return i;
}
}

因为我已经指定了 15 秒的超时时间，所以这是我得到的输出：

pool-2-thread-1 - -764
pool-2-thread-6 - -744
pool-2-thread-2 - 97
pool-2-thread-4 - -563
pool-2-thread-1 - -763
pool-2-thread-6 - -743
pool-2-thread-5 - -463
Exception in thread "main" java.util.concurrent.CancellationException
    at java.util.concurrent.FutureTask$Sync.innerGet(FutureTask.java:220)
    at java.util.concurrent.FutureTask.get(FutureTask.java:83)
    at CallableTest.main(CallableTest.java:44)

如您所见，线程 3 已完成。我想要做的是，在超时期限结束时，如果任何线程到那时还没有完成，我想取消这些线程并设置错误状态，但不要一直抛出异常。我如何做到这一点？

另外，我想显示所有执行的线程和未执行的线程的结果。

由于某种原因，答案被删除了。请把它们留在那里，它可能会帮助那些不完全在寻找这个的人。

Answer 1

实现ExecutorServices 的线程池不提供对其工作线程的访问。因此，请使用自定义ThreadFactory。让它将其线程存储在一个集合中，以便您以后可以中断它们。确保您的工作Runnables 具有适当的中断处理以设置错误状态。提供一个可调用的返回类型，既可以表达错误状态，也可以表达实际的返回值。

Answer 2

我终于弄清楚了如何检查线程是否完成，以及如何处理取消的线程。下面是代码。

public class CallableTest
{
public static void main(String args[]) throws Exception
{
    ExecutorService executor = Executors.newFixedThreadPool(10);
    ExecutorService pool= Executors.newFixedThreadPool(10);
    List<Future<Long>> futureList = new ArrayList<Future<Long>>();
    Set<CallableDemo> callList = new HashSet<CallableDemo>();

         //submit Callable tasks to be executed by thread pool
         //<Long> future = executor.submit(callable);
         //add Future to the list, we can get return value using Future
         //list.add(future);

         callList.add(new CallableDemo(0L));
         callList.add(new CallableDemo(10L));
         callList.add(new CallableDemo(20L));
         callList.add(new CallableDemo(30L));
         callList.add(new CallableDemo(40L));
         callList.add(new CallableDemo(50L));
         callList.add(new CallableDemo(-600L));
         callList.add(new CallableDemo(-700L));
         callList.add(new CallableDemo(-800L));
         callList.add(new CallableDemo(-900L));

        futureList = pool.invokeAll(callList, 15L, TimeUnit.SECONDS);

        for(Future<Long> fut : futureList){
                try {
                    //print the return value of Future, notice the output delay in console
                    // because Future.get() waits for task to get completed
                    if( !fut.isCancelled())
                        System.out.println(new Date()+ "::"+fut.get());
                } catch (InterruptedException e) {
                    //e.printStackTrace();
                    System.out.println("Done :)");
                    //Thread.currentThread().interrupt();
                }
                catch (ExecutionException e) {
                    //e.printStackTrace();
                    System.out.println("Done :)");
                    //Thread.currentThread().interrupt();
                }
            }
            //shut down the executor service now
            executor.shutdown();
            System.out.println("Done :)");
}
}

我使用Future方法isCancelled()，或者你也可以使用isDone()，http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Future.html

public class CallableDemo implements Callable<Long>
{
 private Long count = 0L;

public CallableDemo(Long i)
{
    this.count = i;
}

public Long call() throws InterruptedException
{
    Long i;
    for( i = this.count; i < 100L; i++)
    {
        try { Thread.sleep(100); }
            catch (InterruptedException e) {
                System.out.println("Interruped " + Thread.currentThread().getName());
                //Thread.currentThread().interrupt();
            return i;
        }
        //System.out.println(Thread.currentThread().getName() + " - " + i);
    }
    System.out.println("Finished " + Thread.currentThread().getName());
    return i;
}
}