Java 8 并行 forEach 进度指示

Question

出于性能原因，我想使用并行 Lambda 流的 forEach 循环来处理 Java 中的 Collection 实例。由于它在后台 Service 中运行，我想使用 updateProgress(double,double) 方法来通知用户当前进度。

为了指示当前进度，我需要一个 Integer 计数器形式的进度指示器。但是，这是不可能的，因为我只能访问 Lambda 表达式中的 final 变量。

代码示例见下文，Collection 只是一个占位符，代表任何可能的 Collection 实例:

int progress = 0;
Collection.parallelStream().forEach(signer -> {
   progress++;
   updateProgress(progress, Collection.size());     
});

我知道我可以使用一个简单的 for 循环来解决这个问题。但是，出于性能原因，最好以这种方式解决它。

有没有人知道一个或多或少的巧妙解决方案？

Answer 1

正如 markspace 所建议的，使用 AtomicInteger 是一个很好的解决方案:

AtomicInteger progress = new AtomicInteger();
Collection.parallelStream().forEach(signer -> {
    progress.incrementAndGet();
    // do some other useful work
});

我不会使用 runLater() 变体，因为您的目标是高性能，如果许多并行线程将生成 JavaFX“runLater”任务，您将再次造成瓶颈...

出于同样的原因，我不会每次都对 ProgressBar 调用更新，而是使用 seaparte JavaFX Timeline 独立于处理线程定期更新进度条。

这是一个完整的代码，比较了 ProgressBar 的顺序处理和并行处理。如果您删除 sleep(1) 并将项目数设置为 1000 万，它仍然会并发且高效地工作...

public class ParallelProgress extends Application {
static class ParallelProgressBar extends ProgressBar {
    AtomicInteger myDoneCount = new AtomicInteger();
    int           myTotalCount;
    Timeline      myWhatcher = new Timeline(new KeyFrame(Duration.millis(10), e -> update()));

    public void update() {
        setProgress(1.0*myDoneCount.get()/myTotalCount);
        if (myDoneCount.get() >= myTotalCount) {
            myWhatcher.stop();
            myTotalCount = 0;
        }
    }

    public boolean isRunning() { return myTotalCount > 0; }

    public void start(int totalCount) {
        myDoneCount.set(0);
        myTotalCount = totalCount;
        setProgress(0.0);
        myWhatcher.setCycleCount(Timeline.INDEFINITE);
        myWhatcher.play();
    }

    public void add(int n) {
        myDoneCount.addAndGet(n);
    }
}

HBox testParallel(HBox box) {
    ArrayList<String> myTexts = new ArrayList<String>();

    for (int i = 1; i < 10000; i++) {
        myTexts.add("At "+System.nanoTime()+" ns");
    }

    Button runp = new Button("parallel");
    Button runs = new Button("sequential");
    ParallelProgressBar progress = new ParallelProgressBar();

    Label result = new Label("-");

    runp.setOnAction(e -> {
        if (progress.isRunning()) return;
        result.setText("...");
        progress.start(myTexts.size());

        new Thread() {
            public void run() {
                long ms = System.currentTimeMillis();
                myTexts.parallelStream().forEach(text -> {
                    progress.add(1);
                    try { Thread.sleep(1);} catch (Exception e1) { }
                });
                Platform.runLater(() -> result.setText(""+(System.currentTimeMillis()-ms)+" ms"));
            }
        }.start();
    });

    runs.setOnAction(e -> {
        if (progress.isRunning()) return;
        result.setText("...");
        progress.start(myTexts.size());
        new Thread() {
            public void run() {
                final long ms = System.currentTimeMillis();
                myTexts.forEach(text -> {
                    progress.add(1);
                    try { Thread.sleep(1);} catch (Exception e1) { }
                });
                Platform.runLater(() -> result.setText(""+(System.currentTimeMillis()-ms)+" ms"));
            }
        }.start();
    });

    box.getChildren().addAll(runp, runs, progress, result);
    return box;
}


@Override
public void start(Stage primaryStage) throws Exception {        
    primaryStage.setTitle("ProgressBar's");

    HBox box = new HBox();
    Scene scene = new Scene(box,400,80,Color.WHITE);
    primaryStage.setScene(scene);

    testParallel(box);

    primaryStage.show();   
}

public static void main(String[] args) { launch(args); }
}