JavaFX：运行并行线程并将整体进度更新到 ProgressBar

Question

我需要将一些工作负载拆分到线程并并行启动它们，因为它们是独立的。我还想用 JavaFx 在 ProgressBar 中显示整体进度。这意味着进度条显示到目前为止每个线程完成的总工作。

为简单起见，我们可以以这个Counter 类为例

public class Counter implements Runnable {
    private int from, to;

    public Counter(int from, int to) {
        this.from = from;
        this.to = to;
    }

    @Override
    public void run() {
        for (int i = from; i < to ; i++) {
            // Do some heavy operation
            // report about progress to the parent
        }

        // exit thread with status of success or failure
    }
}

这个类将 from, to 作为边界条件。

为了不阻塞 UI，我使用了一个简单的 Task 类，像这样

public class MyTask extends Task<Integer> {
    int iter;

    public MyTask(int iter) {
        this.iter = iter;
    }

    @Override
    protected Integer call() throws Exception {
        // Simply divide work load to each thread
        int workers = 8;
        int limit = iter/workers;
        int rem = iter % workers;

        // Creates a executor
        ExecutorService executorService = Executors.newFixedThreadPool(workers);
        for (int i = 0; i < workers; i++) {
            int start = limit * i;
            int end = limit * (i + 1);
            if (i == workers - 1) end += rem;

            Counter counter = new Counter(start, end);
            executorService.submit(counter); // Submit work to be done
        }

        executorService.shutdown(); // start the execution

        // Get their progress, update overall progress to UI
        // Stop after all threads finished
    }
}

在MyTask 中，我想按照 cmets 中的说明更新 UI 并整体完成。 （即每个线程完成的总计数）。

有什么方法可以做到这一点吗？聚合并行任务的进度并更新 UI 中的整体进度（我不计算已完成线程的数量，这是我要向 MyTask 报告的每个线程的当前进度）。

Answer 1

根据并行运行的任务数量，您可以简单地将Tasks 用于Counter 逻辑，并将从侦听器的整体进度更新到这些任务的progress 属性。

但是，如果这些任务中有太多并行运行，这可能会减慢 JavaFX 应用程序线程，因为太多的Runnables 可能正在等待一次执行。

您可以使用synchronized 语句自己根据进度差异实施更新。为简单起见，以下代码从 javafx 应用程序线程开始更新，但此逻辑可以移动到不同的线程而不会导致任何问题：

@Override
public void start(Stage primaryStage) {
    ProgressBar progressBar = new ProgressBar();

    int workers = 8;

    ExecutorService executorService = Executors.newFixedThreadPool(workers);

    final int taskCount = 12;
    final int elementsPerTask = 50;
    final int elementCount = elementsPerTask * taskCount;

    ProgressReceiver progressReceiver = new ProgressReceiver() {

        private boolean updating = false;
        private int progress = 0;

        @Override
        public void acceptProgress(int oldValue, int newValue) {
            synchronized(this) {
                progress += newValue - oldValue;
                if (!updating) {
                    updating = true;
                    Platform.runLater(() -> {
                        synchronized (this) {
                            updating = false;
                            progressBar.setProgress(((double) progress) / elementCount);
                        }
                    });
                }
            }
        }

    };
    for (int i = 0; i < taskCount; i++) {
        int start = elementsPerTask * i;
        int end = elementsPerTask * (i + 1);

        Counter counter = new Counter(start, end, progressReceiver);
        executorService.submit(counter);
    }

    executorService.shutdown();

    StackPane root = new StackPane(progressBar);

    Scene scene = new Scene(root, 300, 300);

    primaryStage.setScene(scene);
    primaryStage.show();
}

public interface ProgressReceiver {
    void acceptProgress(int oldValue, int newValue);
}

public class Counter implements Runnable {

    private final ProgressReceiver progressReceiver;

    private final int from, to;

    public Counter(int from, int to, ProgressReceiver progressReceiver) {
        this.from = from;
        this.to = to;
        this.progressReceiver = progressReceiver;
    }

    @Override
    public void run() {
        for (int i = from; i < to; i++) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
            }
            int oldProgress = i - from;
            progressReceiver.acceptProgress(oldProgress, oldProgress + 1);
        }

        // exit thread with status of success or failure
    }
}

Answer 2

我使用PropertyChangeSupport 类解决了这个问题。这提供了线程安全的属性，也提供了属性监听器。更多来自here。

import java.beans.PropertyChangeListener;
import java.beans.PropertyChangeSupport;

public final class ProgressReporter {
    private PropertyChangeSupport propertyChangeSupport = new PropertyChangeSupport(this);
    private int progress = 0;

    public void addPropertyChangeListener(PropertyChangeListener listener) {
        propertyChangeSupport.addPropertyChangeListener(listener);
    }

    public void removePropertyChangeListener(PropertyChangeListener listener) {
        propertyChangeSupport.removePropertyChangeListener(listener);
    }

    public void accumulateProgress(int progress){
        this.propertyChangeSupport.firePropertyChange("progress", this.progress, this.progress + progress);
        this.progress += progress;
    }

    public int getProgress() {
        return progress;
    }
}

现在通过收听ProgressReporter，我们可以了解新数据到达时的进度。请注意，firePropertyChange 仅在 old 和 new 值不同 时触发，否则不会触发对侦听器的更新。

现在我们创建Counter 类来使用这个ProgressReporter

public class Counter implements Runnable {
    private int id, from, to, sleep;
    private ProgressReporter reporter;

    public Counter(int id, int from, int to, int sleep, ProgressReporter reporter) {
        this.from = from;
        this.to = to;
        this.sleep = sleep;
        this.id = id;
        this.reporter = reporter;

        System.out.println("Thread #" + id + " started delay=" + sleep);
    }

    @Override
    public void run() {
        for (int i = from; i < to ; i++) {
            try {
                Thread.sleep(sleep);
                reporter.accumulateProgress(1); // this will fire an update to listeners
            } catch (InterruptedException e){

            }
        }

        System.out.println("Thread #" + id + " is completed");
    }
}

现在在这样实现的 JavaFX 线程发起的任务中。

public class MyTask extends Task<Integer> {
    int iterations;
    Random random = new Random();
    ProgressReporter reporter = new ProgressReporter();

    public MyTask(int iterations) {
        this.iterations = iterations;
    }

    @Override
    protected Integer call() throws Exception {
        // Simply divide work load to each thread
        int workers = 8;
        int limit = iterations /workers;
        int rem = iterations % workers;

        // add a property listener for progress update
        reporter.addPropertyChangeListener(new PropertyChangeListener() {
            @Override
            public void propertyChange(PropertyChangeEvent evt) {
                updateProgress((int) evt.getNewValue(), iterations);
            }
        });

        // Creates a executor
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < workers; i++) {
            int start = limit * i;
            int end = limit * (i + 1);
            if (i == workers - 1) end += rem;

            Counter counter = new Counter(i ,start, end, random.nextInt(1000), reporter);
            executorService.submit(counter); // Submit work to be done
        }

        executorService.shutdown(); // shutdown executor not to accept any more threads
        while (!executorService.isTerminated()){
            if (isCancelled()){
                executorService.shutdownNow(); // stop all the processes immediately
            }
        }

        return reporter.getProgress();
    }
}

现在常见的 JavaFX 绑定如

progressBar.progressProperty().bind(task.progressProperty())

完整的源代码可以在here找到。