如何扩展 FutureTask 并确保释放对 Callable 的引用？

Question

我有一个自定义的 ExecutorService，其中包含一个 ScheduledExecutorService，如果它们花费的时间太长，可以用来中断提交给 ExecutorSerice 的任务，我将完成类放在这篇文章的末尾。

这一切正常，只是有时中断本身会导致问题，所以我将一个 volatile boolean cancel 标志添加到一个新的 CanceableTask 类并使其子类化，以便他们可以检查如果布尔值已被发送为真，则完全停止自己。请注意，它们是提交给执行器服务 precisley 的每个类中的一个布尔实例，因此可以取消长时间运行的任务而无需取消其他任务。

但是 FutureTask 作为参数传递给 beforeExecute(Thread t, Runnable r) 并且这不能访问 Callable 类，所以我的超时代码不能设置取消标志。

我通过重写 newTaskFor 方法来解决这个问题，返回一个只提供对 Callable 的引用的类

public class FutureCallable<V> extends FutureTask<V>
{
    private Callable<V> callable;
    public FutureCallable(Callable<V> callable) {
        super(callable);
        this.callable = callable;
    }
    public Callable<V> getCallable() {
        return callable;
    }
}

一切都很好，至少我是这么认为的。

不幸的是，我的应用程序现在使用越来越多的内存，因为新任务被提交给 ExecutorService 并最终耗尽内存，当我分析应用程序时，我发现有一个对所有 FutureCallables 的线程堆栈本地引用，即使在之后任务已经完成，因为 FutureCallable 引用了正在运行的类，所以它使用了大量内存。

当我查看（FutureCallable 扩展）的 FutureTask 代码时，有一条关于私有 Callable 引用的注释，上面写着

/** The underlying callable; nulled out after running */

那么我该如何改进我的 FutureCallable 以取消其对 Callable 的引用？ 或者为什么在任务完成后维护对 FutureCallable 的引用。

我已经确认如果我注释掉 newTaskFor 方法没有过多的内存使用，但不幸的是我无法取消课程。

完整的类是：

public class TimeoutThreadPoolExecutor extends ThreadPoolExecutor {
    private final long timeout;
    private final TimeUnit timeoutUnit;

    private final static int WAIT_BEFORE_INTERRUPT = 10000;
    private final static int WAIT_BEFORE_STOP      = 10000;


    private final ScheduledExecutorService timeoutExecutor = Executors.newSingleThreadScheduledExecutor();

    //Map Task to the Future of the Timeout Task that could be used to interrupt it
    private final ConcurrentMap<Runnable, ScheduledFuture> runningTasks = new ConcurrentHashMap<Runnable, ScheduledFuture>();

    public long getTimeout()
    {
        return timeout;
    }

    public TimeUnit getTimeoutUnit()
    {
        return timeoutUnit;
    }

    public TimeoutThreadPoolExecutor(int workerSize, ThreadFactory threadFactory, long timeout, TimeUnit timeoutUnit)
    {
        super(workerSize, workerSize, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), threadFactory);
        MainWindow.logger.severe("Init:"+workerSize+":Timeout:"+timeout+":"+timeoutUnit);
        this.timeout = timeout;
        this.timeoutUnit = timeoutUnit;
    }

    public TimeoutThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, long timeout, TimeUnit timeoutUnit) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        this.timeout = timeout;
        this.timeoutUnit = timeoutUnit;
    }

    @Override
    public <T> FutureCallable<T> newTaskFor(Callable<T> callable) {
        return new FutureCallable<T>(callable);
    }

    @Override
    public List<Runnable> shutdownNow() {
        timeoutExecutor.shutdownNow();
        return super.shutdownNow();
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        if(timeout > 0) {
            //Schedule a task to interrupt the thread that is running the task after time timeout starting from now
            final ScheduledFuture<?> scheduled = timeoutExecutor.schedule(new TimeoutTask(t, r), timeout, timeoutUnit);

            //Add Mapping
            runningTasks.put(r, scheduled);
        }
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {

        //AfterExecute will be called after the task has completed, either of its own accord or because it
        //took too long and was interrupted by corresponding timeout task
        //Remove mapping and cancel timeout task
        ScheduledFuture timeoutTask = runningTasks.remove(r);
        if(timeoutTask != null) {
            timeoutTask.cancel(false);
        }

    }

    @Override
    protected void terminated()
    {
        //All tasks have completed either naturally or via being cancelled by timeout task so close the timeout task
        MainWindow.logger.severe("---Shutdown TimeoutExecutor");
        timeoutExecutor.shutdown();
    }

    /**
     * Interrupt or possibly stop the thread
     *
     */
    class TimeoutTask implements Runnable {
        private final       Thread thread;
        private             Callable c;

        public TimeoutTask(Thread thread, Runnable c) {
            this.thread = thread;
            if(c instanceof FutureCallable)
            {
                this.c = ((FutureCallable) c).getCallable();
            }
        }

        @Override
        public void run()
        {
            String msg = "";
            if (c != null)
            {
                if (c != null && c instanceof CancelableTask)
                {
                    MainWindow.logger.severe("+++Cancelling " + msg + " task because taking too long");
                    ((CancelableTask) c).setCancelTask(true);
                }
            }
        }
    }
}

    public abstract class CancelableTask  extends ExecutorServiceEnabledAnalyser
    {
        private volatile boolean cancelTask = false;

        public boolean isCancelTask() {
            return cancelTask;
        }

        public void setCancelTask(boolean cancelTask) {
            this.cancelTask = cancelTask;
        }

        CancelableTask(final MainWindow start, boolean isSelectedRecords, boolean isUseRowSelection)
        {
            super(start, isSelectedRecords, isUseRowSelection);
        }

        CancelableTask(final MainWindow start, List<MetadataChangedWrapper> songs)
        {
            super(start, songs );
        }

    }

Answer 1

这是ThreadLocal 在哪里？我觉得很奇怪，很难相信你在说什么，它保留了对所有运行过的任务的引用，即使在完成之后也是如此。如果是这种情况，即使没有您的覆盖，它最终也会耗尽内存（尽管任务本身使用了一些内存，但可能少于您的可调用内存，但仍然不为零）。

无论如何，您可以覆盖 FutureCallable 上的 done 方法，以在执行后将包装的对象清空。