使用 ThreadPoolExecutor 取消 SwingWorker

Question

我正在使用线程池大小为 1 的 ThreadPoolExecutor 来顺序执行 swing 工作线程。 我遇到了一个特殊情况，其中一个事件到达，创建了一个 Swing Worker，它执行一些客户端-服务器通信，然后更新 ui（在 did() 方法中）。

当用户触发（单击某个项目）某些事件时，此方法可以正常工作，但如果发生很多事件，则效果不佳。 但发生这种情况，所以我需要取消所有当前正在运行和计划的工作人员。 问题是支持 ThreadPoolExecutor 的队列不知道 SwingWorker 取消过程（至少看起来是这样）。 因此，预定的工作人员会被取消，但已经运行的工作人员不会被取消。

因此，我添加了一个 类型的并发队列，它保存所有工作人员的引用，只要它们没有被取消，并且当新事件到达时，它会对所有工作人员调用 .cancel(true)将 SwingWorker 放入队列中并将新的 SwingWorker 提交给 ThreadPoolExecutor。

摘要： SwingWorkers 在具有单个线程的 ThreadPoolExecutor 中创建和执行。 只有最后提交的工作人员才应该运行。

有没有其他方法可以解决这个问题，或者这样做“可以”吗？

只是好奇...

Answer 1

为什么需要 ThreadPoolExecutor 来完成这种工作？

您有多少个不同 SwingWorkers 的来源？ 因为如果来源只是一个，您应该使用不同的方法。

例如，您可以定义一个处理一种工作线程的类，并将其链接到一种项目，用户可以在该项目上触发操作并在该类中关心应该运行线程的单个实例的事实（例如使用完成任务后清除的单例实例）

Answer 2

您是否可以不使用 SwingWorker 来使用 ThreadPoolExecutor 来执行客户端-服务器通信，然后调用 SwingUtilities.invokeLater 来使用结果更新 UI？ 对我来说，这似乎更干净一些，并且可以确保事件和 UI 更新仍然按顺序处理。

当您向执行者提交任务时，您可以保留对其 Future 实例的引用，以便您可以在需要时取消任务。

Answer 3

让我看看我是否正确理解了这个问题。 您有一个 FIFO 任务队列，只有最旧的任务正在运行。 每个任务完成后都需要更新 UI。 但是，如果某个用户事件到来，则所有任务都需要取消——也就是说，正在运行的任务需要取消，尚未运行的任务需要从队列中删除。 是对的吗？

假设是这样，我不会使用 SwingWorker 因为您只需要一个工作线程，而不是每个任务一个。 FutureTask 应该足够了（假设您重写 done() 以对 SwingUtilities.invokeLater() 进行必要的调用并进行 UI 更新）。

如果您取消 FutureTask，那么即使它的 run() 方法被调用，它也不会执行任何操作。 因此，您可以安全地将 FutureTask 提交给 ExecutorService，因为知道即使执行程序尝试运行它们，取消也会起作用。

我怀疑一个足够好的解决方案就是保留可能需要取消的所有 FutureTasks 的列表，并在用户事件到来时将它们全部取消。ExecutorService code> 仍会尝试运行它们，但基本上是无操作。 您需要确保已完成的任务从列表中删除，并且需要确保列表以线程安全的方式更新和使用（可能来自将任务放在 ExecutorService 上的同一线程） ），但这应该不会太难。

我在短短一个小时内就把下面的代码敲出来了，我不敢打赌它是正确的，但你明白了。 :)

/** Untested code! Use at own risk. */
public class SwingTaskExecutor {

    // ////////////////////////////////////////////////////////////
    // Fields

    private final ExecutorService execSvc = Executors.newFixedThreadPool(1);

    private final Lock listLock = new ReentrantLock();
    private final List<ManagedSwingTask<?>> activeTasks = 
            new ArrayList<ManagedSwingTask<?>>();

    // ////////////////////////////////////////////////////////////
    // Public methods

    public <T> Future<T> submit(SwingTask<T> task) {
        ManagedSwingTask<T> managedTask = new ManagedSwingTask<T>(task);
        addToActiveTasks(managedTask);
        execSvc.submit(managedTask);
        return managedTask;
    }

    public void cancelAllTasks() {
        listLock.lock();
        try {
            for (ManagedSwingTask<?> t: activeTasks) {
                t.cancel(true);
            }
            activeTasks.clear();
        } finally {
            listLock.unlock();
        }
    }

    // ////////////////////////////////////////////////////////////
    // Private methods

    private <T> void addToActiveTasks(ManagedSwingTask<T> managedTask) {
        listLock.lock();
        try {
            activeTasks.add(managedTask);
        } finally {
            listLock.unlock();
        }
    }

    // ////////////////////////////////////////////////////////////
    // Helper classes

    private class ManagedSwingTask<T> extends FutureTask<T> {

        private final SwingTask<T> task;

        ManagedSwingTask(SwingTask<T> task) {
            super(task);
            this.task = task;
        }

        @Override
        public void cancel(boolean mayInterruptIfRunning) {
            try {
                task.cancel();
            } finally {
                super.cancel(mayInterruptIfRunning);
            }
        }

        @Override
        protected void done() {
            removeFromActiveTasks();
            updateUIIfDone();
        }

        private void removeFromActiveTasks() {
            listLock.lock();
            try {
                activeTasks.remove(this);
            } finally {
                listLock.unlock();
            }
        }

        private void updateUIIfDone() {
            if (isDone()) {
                SwingUtilities.invokeLater(new Runnable() {
                    @Override
                    public void run() {
                        task.updateUI();
                    }
                });
            }
        }
    }

    public static interface SwingTask<T> extends Callable<T> {

        /** Called from the EDT if task completes successfully */
        void updateUI();

        /** Hook in case there's task-specific cancellation to be done*/
        void cancel();
    }
}

无论如何，类似的事情。

如果您想更加确定，则可以关闭并替换 ExecutorService，但这可能没有必要。

Answer 4

创建仅执行最后传入的 Runnable 的单线程 ThreadPoolExecutor 的一种方法是子类化合适的队列类并重写所有添加方法以在添加新的可运行之前清除队列。 然后将该队列设置为 ThreadPoolExecutor 的工作队列。