同步异步调用序列

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