同步一系列异步调用

Question

我正在使用 JavaFX 的 WebView 来解析网站。该站点包含一堆链接 - 我需要按照给定的顺序分别打开每个链接，并从每个链接中检索一个信息。

为了确保WebView 已加载整个站点，我正在监听WebEngine 的changed 事件并等待newState == Worker.State.SUCCEEDED。问题是这个调用是异步的。当我调用webEngine.load(firstAddress); 时，代码立即返回，在此页面加载之前，我的代码将调用另一个webEngine.load(secondAddress);，依此类推。

我明白为什么要这样做（为什么异步比同步好），但我是 Java 初学者，我不确定解决这个问题的最佳方法是什么。我不知何故了解多线程和东西，所以我已经尝试过信号量（CountDownLatch 类）。但是代码挂在await 上，我不确定我做错了什么。

有人可以告诉我应该如何以正确的方式完成吗？也许一些通用模式如何应对这样的场景？

我想要实现的伪代码：

WebEngine webEngine = new WebEngine();
webEngine.loadPage("http://www.something.com/list-of-cars");
webEngine.waitForThePageToLoad(); // I need an equivalent of this. In the real code, this is done asynchronously as a callback
// ... some HTML parsing or DOM traversing ...
List<String> allCarsOnTheWebsite = webEngine.getDocument()....getChildNodes()...;
// allCarsOnTheWebsite contains URLs to the pages I want to analyze

for (String url : allCarsOnTheWebsite)
{
    webEngine.loadPage(url);
    webEngine.waitForThePageToLoad(); // same as in line 3

    String someDataImInterestedIn = webEngine.getDocument()....getChildNodes()...Value();
    System.out.println(url + " : " + someDataImInterestedIn);
}

System.out.println("Done, all cars have been analyzed");

Answer 1

您应该使用在页面加载时调用的侦听器，而不是在加载完成之前阻塞。

类似：

WebEngine webEngine = new WebEngine();
ChangeListener<State> initialListener = new ChangeListener<State>() {
    @Override
    public void changed(ObservableValue<? extends State> obs, State oldState, State newState) {
        if (newState == State.SUCCEEDED) {
            webEngine.getLoadWorker().stateProperty().removeListener(this);
            List<String> allCarsOnTheWebsite = webEngine.getDocument()... ;
            loadPagesConsecutively(allCarsOnTheWebsite, webEngine);
        }
    }
};
webEngine.getLoadWorker().addListener(initialListener);      
webEngine.loadPage("http://www.something.com/list-of-cars");

// ...

private void loadPagesConsecutively(List<String> pages, WebEngine webEngine) {
    LinkedList<String> pageStack = new LinkedList<>(pages);
    ChangeListener<State> nextPageListener = new ChangeListener<State>() {
        @Override
        public void changed(ObservableValue<? extends State> obs, State oldState, State newState) {
            if (newState == State.SUCCEEDED ) {
                // process current page data
                // ...
                if (pageStack.isEmpty()) {
                    webEngine.getLoadWorker().stateProperty().removeListener(this);
                } else {
                    // load next page:
                    webEngine.load(pageStack.pop());
                }
            }               
        }
    };
    webEngine.getLoadWorker().stateProperty().addListener(nextPageListener);

    // load first page (assumes pages is not empty):
    webEngine.load(pageStack.pop());
}

Answer 2

如果您想同时运行所有任务，但要按照提交的顺序处理它们，请查看以下示例：

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javafx.application.Application;
import javafx.application.Platform;
import javafx.concurrent.Task;
import javafx.scene.Scene;
import javafx.scene.control.ListView;
import javafx.scene.layout.BorderPane;
import javafx.stage.Stage;

public class ProcessTaskResultsSequentially extends Application {

    @Override
    public void start(Stage primaryStage) {
        ListView<String> results = new ListView<>();

        List<Task<Integer>> taskList = new ArrayList<>();
        for (int i = 1; i<= 10 ; i++) {
            taskList.add(new SimpleTask(i));
        }

        ExecutorService exec = Executors.newCachedThreadPool(r -> {
            Thread t = new Thread(r);
            t.setDaemon(true);
            return t ;
        });


        Thread processThread = new Thread(() -> {
            for (Task<Integer> task : taskList) {
                try {
                    int result = task.get();
                    Platform.runLater(() -> {
                        results.getItems().add("Result: "+result);
                    });
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        processThread.setDaemon(true);
        processThread.start();

        taskList.forEach(exec::submit);

        primaryStage.setScene(new Scene(new BorderPane(results), 250, 400));
        primaryStage.show();
    }

    public static class SimpleTask extends Task<Integer> {
        private final int index ;

        private final static Random rng = new Random();

        public SimpleTask(int index) {
            this.index = index ;
        }

        @Override
        public Integer call() throws Exception {
            System.out.println("Task "+index+" called");
            Thread.sleep(rng.nextInt(1000)+1000);
            System.out.println("Task "+index+" finished");
            return index ;
        }
    }

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