java - ExecutorService 是否有一个解决方案，可以限制每个提交的 Callable 的时间

Question

这就是我所拥有的...我有一个任务列表，每个任务都将在其自己的 ExecutorService 上提交。我希望每个任务在开始执行后都有一个 x MS 的时间限制。我见过像 Time Limit on Individual Threads 上发布的解决方案但这设置了从提交任务时开始的时间限制。

我认为一种解决方案是将已接受的解决方案扩展到 Time Limit on Individual Threads 。这会将 Callable 包装在另一个 Callable 中，一旦启动，就会将取消任务置于计划的执行程序上。这将涉及到以一种不平凡的方式将Callable映射到Future，所以我想我会发帖看看是否有人知道现有的解决方案(代码重新)使用成为一件美妙的事情等等)。

谢谢。

仅供引用，解决方案发布于 Time Limit on Individual Threads如下。同样，这会从提交时间取消，而不是从开始时间取消:

ExecutorService service = Executors.newFixedThreadPool(N);
ScheduledExecutorService canceller = Executors.newSingleThreadScheduledExecutor();

public <T> Future<T> executeTask(Callable<T> c, long timeoutMS){
   final Future<T> future = service.submit(c);
   canceller.schedule(new Callable<Void>(){
       public Void call(){
          future.cancel(true);
          return null;
       }
    }, timeoutMS, TimeUnit.MILLI_SECONDS);
   return future;
}

Answer 1

看看这是否适合您。

抱歉，如果代码很困惑，但这只是为了演示这个概念:将 future 传回任务，让任务启动自己的计时器 - 计时器将在任务启动时启动。

以下代码添加了 10 个任务，每个任务需要 2 秒执行，但超时时间为 1 秒。

import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class CancellableTaskDemo {
    public static void main(final String[] args) {
        new CancellableTaskDemo();
    }

    final ExecutorService ex = Executors.newFixedThreadPool(3);

    public CancellableTaskDemo() {
        for (int i = 0; i < 10; i++) {
            final int c = i;
            submitTask(new Callable<Object>() {
                @Override
                public Object call() throws Exception {
                    final long t = System.currentTimeMillis();
                    try {
                        Thread.sleep(2000);
                        System.out.println("Task " + c + " done in " + (System.currentTimeMillis() - t) + "ms");
                    } catch (final InterruptedException e) {
                        System.out.println("Task " + c + " aborted after " + (System.currentTimeMillis() - t) + "ms");
                    }
                    return null;
                }
            }, 1000);
        }
        ex.shutdown();
        try {
            ex.awaitTermination(100000, TimeUnit.MILLISECONDS);
        } catch (final InterruptedException e) {
            e.printStackTrace();
        }
    }

    private void submitTask(final Callable<Object> c, final int timeout) {
        final TimedFutureTask tft = new TimedFutureTask(c, timeout);
        final Future<?> ft = ex.submit(tft.getCallable());
        tft.setFuture(ft);
    }

    static class TimedFutureTask {
        final static ScheduledExecutorService canceller = Executors.newSingleThreadScheduledExecutor();
        private Timer cancelTimer;
        private Callable<Object> timedCallable;
        private Future<?> f;
        private int timeoutMS;

        public TimedFutureTask(final Callable<Object> callable, final int timeoutMS) {
            this.timeoutMS = timeoutMS;
            timedCallable = (new Callable<Object>() {

                @Override
                public Object call() throws Exception {

                    cancelTimer = new Timer();
                    cancelTimer.schedule(new TimerTask() {

                        @Override
                        public void run() {
                            f.cancel(true);

                        }
                    }, timeoutMS);

                    final Object res = callable.call();
                    cancelTimer.cancel();
                    return res;
                }

            });
        }

        public Callable<Object> getCallable() {
            return timedCallable;
        }

        public void setFuture(final Future<?> future) {
            f = future;
        }
    }

}