Java 8 parallel forEach进度指示 [英] Java 8 parallel forEach progress indication

问题描述

出于性能原因，我想使用一个并行Lambda流的forEach循环，以便在Java中处理一个 Collection 的实例。因为它在后台运行 Service 我想使用 updateProgress（double，double）方法

For performance reason I would like to use a forEach loop of a parallel Lambda stream in order to process an instance of a Collection in Java. As this runs in a background Service I would like to use the updateProgress(double,double) method in order to inform the user about the current progress.

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

In order to indicate the current progress I need a certain progress indicator in form of a Integer counter. However, this is not possible as I can only access final variables within the Lambda expression.

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

Code example see below, Collection is only a place holder for any possible instance of a Collection:

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

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

I'm aware that I can solve this problem by using a simple for-loop. However, for performance reason it would nice to solve it in this way.

有人知道一个或多或少的整洁的解决方案吗？

Does anybody know a more or less neat solution to this?

推荐答案

根据markspace的建议，使用AtomicInteger是一个很好的解决方案：

As proposed by markspace, using an AtomicInteger is a good solution:

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

我不会使用runLater（）变体作为你的目标是一个高性能，如果许多并行线程将生成JavaFX的'runLater'任务，你将再次创建一个瓶颈...

I would not use the runLater() variant as your goal is a high performance, and if many parallel threads will generte JavaFX 'runLater' tasks, you will again create a bottleneck...

由于同样的原因，我不会每次调用ProgressBar的更新，使用seaparte JavaFX时间轴以固定的时间间隔独立于处理线程更新进度条。

For the same reason I would NOT call an update to the ProgressBar each time, but use a seaparte JavaFX Timeline to update the progress bar in regular intervals independently from the processing threads.

这里是一个完整的代码，比较了ProgressBar的顺序和并行处理。如果你删除sleep（1）并将项目数设置为1000万，它仍然可以同时有效地工作...

Here is a full code comparing sequential and parallel processing with ProgressBar. If you remove the sleep(1) and set the number of items to 10 million it will still work concurrently and efficiently...

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); }
}

这篇关于Java 8 parallel forEach进度指示的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！