SwingWorker, done() 在 process() 调用完成之前执行

Question

我已经和SwingWorkers 合作了一段时间，结果出现了一种奇怪的行为，至少对我来说是这样。我清楚地明白，由于性能原因，对publish() 方法的多次调用在一次调用中合并。这对我来说非常有意义，我怀疑 SwingWorker 会保留某种队列来处理所有调用。

根据tutorial和API，当SwingWorker结束执行时，要么doInBackground()正常结束要么从外部取消工作线程，然后调用done()方法。到目前为止一切顺利。

但我有一个示例（类似于教程中所示），其中有process() 方法调用完成 done() 方法被执行。由于这两种方法都在Event Dispatch Thread 中执行，我希望done() 在所有process() 调用完成后执行。换句话说：

预期：

Writing...
Writing...
Stopped!

结果：

Writing...
Stopped!
Writing...

示例代码

import java.awt.BorderLayout;
import java.awt.Dimension;
import java.awt.Graphics;
import java.awt.event.ActionEvent;
import java.util.List;
import javax.swing.AbstractAction;
import javax.swing.Action;
import javax.swing.JButton;
import javax.swing.JFrame;
import javax.swing.JPanel;
import javax.swing.JScrollPane;
import javax.swing.JTextArea;
import javax.swing.SwingUtilities;
import javax.swing.SwingWorker;

public class Demo {

    private SwingWorker<Void, String> worker;
    private JTextArea textArea;
    private Action startAction, stopAction;

    private void createAndShowGui() {

        startAction = new AbstractAction("Start writing") {
            @Override
            public void actionPerformed(ActionEvent e) {
                Demo.this.startWriting();
                this.setEnabled(false);
                stopAction.setEnabled(true);
            }
        };

        stopAction = new AbstractAction("Stop writing") {
            @Override
            public void actionPerformed(ActionEvent e) {
                Demo.this.stopWriting();
                this.setEnabled(false);
                startAction.setEnabled(true);
            }
        };

        JPanel buttonsPanel = new JPanel();
        buttonsPanel.add(new JButton(startAction));
        buttonsPanel.add(new JButton(stopAction));

        textArea = new JTextArea(30, 50);
        JScrollPane scrollPane = new JScrollPane(textArea);

        JFrame frame = new JFrame("Test");
        frame.setDefaultCloseOperation(JFrame.DISPOSE_ON_CLOSE);
        frame.add(scrollPane);
        frame.add(buttonsPanel, BorderLayout.SOUTH);
        frame.pack();
        frame.setLocationRelativeTo(null);
        frame.setVisible(true);
    }

    private void startWriting() {
        stopWriting();
        worker = new SwingWorker<Void, String>() {
            @Override
            protected Void doInBackground() throws Exception {
                while(!isCancelled()) {
                    publish("Writing...\n");
                }
                return null;
            }

            @Override
            protected void process(List<String> chunks) {
                String string = chunks.get(chunks.size() - 1);
                textArea.append(string);
            }

            @Override
            protected void done() {
                textArea.append("Stopped!\n");
            }
        };
        worker.execute();
    }

    private void stopWriting() {
        if(worker != null && !worker.isCancelled()) {
            worker.cancel(true);
        }
    }

    public static void main(String[] args) {
        SwingUtilities.invokeLater(new Runnable() {
            @Override
            public void run() {
                new Demo().createAndShowGui();
            }
        });
    }
}

Answer 1

在阅读了 DSquare 的出色答案并从中得出结论认为需要进行一些子类化后，我提出了这个想法，供任何需要确保所有已发布的块在继续之前在 EDT 中处理过的人。

注意，我尝试用 Java 而不是 Jython（我选择的语言，官方称它是世界上最好的语言）编写它，但它有点复杂，因为例如，publish 是 final，所以你必须发明另一种方法来调用它，而且还因为您必须（打哈欠）使用 Java 中的泛型参数化所有内容。

任何 Java 人员都应该可以理解这段代码：只是为了提供帮助，使用 self.publication_counter.get()，当结果为 0 时，它的计算结果为 False。

# this is how you say Worker... is a subclass of SwingWorker in Python/Jython
class WorkerAbleToWaitForPublicationToFinish( javax.swing.SwingWorker ):

    # __init__ is the constructor method in Python/Jython
    def __init__( self ):

        # we just add an "attribute" (here, publication_counter) to the object being created (self) to create a field of the new object
        self.publication_counter = java.util.concurrent.atomic.AtomicInteger()

    def await_processing_of_all_chunks( self ):
        while self.publication_counter.get():
            time.sleep( 0.001 )

    # fully functional override of the Java method     
    def process( self, chunks ):
        for chunk in chunks:
            pass
            # DO SOMETHING WITH EACH CHUNK

        # decrement the counter by the number of chunks received
        # NB do this AFTER dealing with the chunks 
        self.publication_counter.addAndGet( - len( chunks ) )

    # fully functional override of the Java method     
    def publish( self, *chunks ):
        # increment the counter by the number of chunks received
        # NB do this BEFORE publishing the chunks
        self.publication_counter.addAndGet( len( chunks ))
        self.super__publish( chunks )

因此，在您的调用代码中，您可以输入以下内容：

    engine.update_xliff_task.get()
    engine.update_xliff_task.await_processing_of_all_chunks()

PS 像这样使用while 子句（即轮询技术）并不优雅。我查看了可用的java.util.concurrent 类，例如CountDownLatch 和Phaser（都具有线程阻塞方法），但我认为它们都不适合此目的......

稍后

我对此非常感兴趣，因此调整了一个适当的并发类（用 Java 编写，可在 Apache 网站上找到），名为 CounterLatch。他们的版本在await() 如果 达到AtomicLong 计数器的值时停止线程。我在这里的版本允许您这样做，或者相反：说“等待直到计数器在解除闩锁之前达到某个值”：

NB 使用AtomicLong 表示signal 和AtomicBoolean 表示released：因为在原始Java 中它们使用volatile 关键字。我认为使用原子类将达到相同的目的。

class CounterLatch():
    def __init__( self, initial = 0, wait_value = 0, lift_on_reached = True ):
        self.count = java.util.concurrent.atomic.AtomicLong( initial )
        self.signal = java.util.concurrent.atomic.AtomicLong( wait_value )

        class Sync( java.util.concurrent.locks.AbstractQueuedSynchronizer ):
            def tryAcquireShared( sync_self, arg ):
                if lift_on_reached:
                    return -1 if (( not self.released.get() ) and self.count.get() != self.signal.get() ) else 1
                else:
                    return -1 if (( not self.released.get() ) and self.count.get() == self.signal.get() ) else 1
            def tryReleaseShared( self, args ):
                return True

        self.sync = Sync()
        self.released = java.util.concurrent.atomic.AtomicBoolean() # initialised at False

    def await( self, *args ):
        if args:
            assert len( args ) == 2
            assert type( args[ 0 ] ) is int
            timeout = args[ 0 ]
            assert type( args[ 1 ] ) is java.util.concurrent.TimeUnit
            unit = args[ 1 ]
            return self.sync.tryAcquireSharedNanos(1, unit.toNanos(timeout))
        else:
            self.sync.acquireSharedInterruptibly( 1 )

    def count_relative( self, n ):
        previous = self.count.addAndGet( n )
        if previous == self.signal.get():
            self.sync.releaseShared( 0 )
        return previous

所以我的代码现在看起来像这样：

在 SwingWorker 构造函数中：

self.publication_counter_latch = CounterLatch() 

在 SW.publish 中：

self.publication_counter_latch.count_relative( len( chunks ) )
self.super__publish( chunks )

在等待块处理停止的线程中：

worker.publication_counter_latch.await()

Answer 2

简短回答：

这是因为 publish() 不直接调度 process，它设置了一个计时器，该计时器将在 DELAY 之后触发 EDT 中的 process() 块的调度。因此，当工作人员被取消时，仍然有一个计时器在等待使用上次发布的数据来安排一个 process()。使用定时器的原因是为了实现一个进程可以执行多个发布的组合数据的优化。

长答案：

让我们看看 publish() 和 cancel 是如何相互作用的，为此，让我们深入研究一些源代码。

首先是简单的部分，cancel(true)：

public final boolean cancel(boolean mayInterruptIfRunning) {
    return future.cancel(mayInterruptIfRunning);
}

此取消最终调用以下代码：

boolean innerCancel(boolean mayInterruptIfRunning) {
    for (;;) {
        int s = getState();
        if (ranOrCancelled(s))
            return false;
        if (compareAndSetState(s, CANCELLED)) // <-----
            break;
    }
    if (mayInterruptIfRunning) {
        Thread r = runner;
        if (r != null)
            r.interrupt(); // <-----
    }
    releaseShared(0);
    done(); // <-----
    return true;
}

SwingWorker状态设置为CANCELLED，线程被中断，调用done()，然而这不是SwingWorker的done，而是futuredone()，在变量实例化时指定SwingWorker 构造函数：

future = new FutureTask<T>(callable) {
    @Override
    protected void done() {
        doneEDT();  // <-----
        setState(StateValue.DONE);
    }
};

doneEDT() 代码是：

private void doneEDT() {
    Runnable doDone =
        new Runnable() {
            public void run() {
                done(); // <-----
            }
        };
    if (SwingUtilities.isEventDispatchThread()) {
        doDone.run(); // <-----
    } else {
        doSubmit.add(doDone);
    }
}

如果我们在 EDT 中，它会直接调用 SwingWorkers 的 done()，这就是我们的情况。此时 SwingWorker 应该停止，不应再调用 publish()，这很容易通过以下修改来演示：

while(!isCancelled()) {
    textArea.append("Calling publish\n");
    publish("Writing...\n");
}

但是，我们仍然从 process() 收到“Writing...”消息。那么让我们看看 process() 是如何被调用的。 publish(...)的源码是

protected final void publish(V... chunks) {
    synchronized (this) {
        if (doProcess == null) {
            doProcess = new AccumulativeRunnable<V>() {
                @Override
                public void run(List<V> args) {
                    process(args); // <-----
                }
                @Override
                protected void submit() {
                    doSubmit.add(this); // <-----
                }
            };
        }
    }
    doProcess.add(chunks);  // <-----
}

我们看到 Runnable doProcess 的 run() 最终调用了process(args)，但这段代码只调用了doProcess.add(chunks) 而不是doProcess.run()，而且还有一个doSubmit。让我们看看doProcess.add(chunks)。

public final synchronized void add(T... args) {
    boolean isSubmitted = true;
    if (arguments == null) {
        isSubmitted = false;
        arguments = new ArrayList<T>();
    }
    Collections.addAll(arguments, args); // <-----
    if (!isSubmitted) { //This is what will make that for multiple publishes only one process is executed
        submit(); // <-----
    }
}

所以 publish() 实际上所做的是将块添加到一些内部 ArrayList arguments 并调用 submit()。我们刚刚看到提交只调用了doSubmit.add(this)，这与add 方法完全相同，因为doProcess 和doSubmit 都扩展了AccumulativeRunnable<V>，但是这次V 是Runnable 而不是@987654356 @ 如doProcess。所以一个块是调用process(args) 的可运行对象。但是submit() 调用是在doSubmit 类中定义的完全不同的方法：

private static class DoSubmitAccumulativeRunnable
     extends AccumulativeRunnable<Runnable> implements ActionListener {
    private final static int DELAY = (int) (1000 / 30);
    @Override
    protected void run(List<Runnable> args) {
        for (Runnable runnable : args) {
            runnable.run();
        }
    }
    @Override
    protected void submit() {
        Timer timer = new Timer(DELAY, this); // <-----
        timer.setRepeats(false);
        timer.start();
    }
    public void actionPerformed(ActionEvent event) {
        run(); // <-----
    }
}

它创建一个计时器，在DELAY 毫秒后触发一次actionPerformed 代码。一旦事件被触发，代码将在 EDT 中排队，这将调用内部 run()，最终调用 doProcess 的 run(flush()) 并因此执行 process(chunk)，其中块是 arguments 的刷新数据数组列表。我跳过了一些细节，“运行”调用链是这样的：

• doSubmit.run()
• doSubmit.run(flush()) //实际上是一个runnables循环，但只会有一个（*）
• doProcess.run()
• doProcess.run(flush())
• 进程（块）

(*)布尔值 isSubmited 和 flush() （重置此布尔值）使得额外的发布调用不会添加要在 doSubmit.run(flush()) 中调用的 doProcess 可运行对象，但它们的数据是不被忽视。因此，为在 Timer 生命周期内调用的任意数量的发布执行单个进程。

总而言之，publish("Writing...") 所做的是在 EDT 在延迟之后安排对 process(chunk) 的调用。这解释了为什么即使在我们取消线程并且没有完成更多发布之后，仍然会出现一个进程执行，因为在我们取消工作人员的那一刻，有（很有可能）一个计时器将在 done() 之后安排一个 process() 已经预定。

为什么在 EDT 中使用此 Timer 而不是仅使用 invokeLater(doProcess) 调度 process()？实现docs中解释的性能优化：

因为在Event上异步调用了process方法 Dispatch Thread 对发布方法的多次调用可能会发生 在执行处理方法之前。出于性能目的，所有 这些调用被合并为一个调用并连接 论据。 例如：

 publish("1");
 publish("2", "3");
 publish("4", "5", "6");

might result in:
 process("1", "2", "3", "4", "5", "6")

我们现在知道这是可行的，因为在 DELAY 间隔内发生的所有发布都将它们的 args 添加到我们看到的内部变量 arguments 中，并且 process(chunk) 将一次性执行所有这些数据。

这是一个错误吗？解决方法？

很难判断这是否是错误，处理后台线程发布的数据可能有意义，因为工作实际上已经完成，您可能有兴趣使用尽可能多的信息更新 GUI尽你所能（例如，如果这就是 process() 正在做的事情）。如果done() 需要处理所有数据和/或在 done() 之后调用 process() 会导致数据/GUI 不一致，那么它可能没有意义。

如果您不想在 done() 之后执行任何新的 process()，有一个明显的解决方法，只需在 process 方法中检查 worker 是否也被取消！

@Override
protected void process(List<String> chunks) {
    if (isCancelled()) return;
    String string = chunks.get(chunks.size() - 1);
    textArea.append(string);
}

在最后一个 process() 之后执行 done() 更加棘手，例如 done 也可以使用一个计时器，该计时器将在 >DELAY 之后安排实际的 done() 工作。虽然我不认为这是一个常见的情况，因为如果你取消了，当我们知道我们实际上正在取消所有未来进程的执行时，再错过一个 process() 应该不重要。