JavaFX 和多线程 : IllegalStateException on ObservableList. add()

Question

我有一个 JavaFX 应用程序，它与服务器通信并从服务器获取数据。接收到的数据放入 ObservableList 中并显示在 TableView 中。

与服务器的通信在其自己的线程中运行，当调用 ObservableList.add 时，它会触发 IllegalStateException(提示不是事件线程/不是 JavaFX 线程)

我发现了以下solution类似的问题，但我不确定如何采用这种方法，因为在我的情况下，与服务器的通信需要不断地保持，因此任务/线程正在运行，直到通信终止。

我这里有一个最小的工作示例，它会触发所述异常并大致模拟应用程序的工作方式。

主要:

package sample;

import javafx.application.Application;
import javafx.fxml.FXMLLoader;
import javafx.scene.Parent;
import javafx.scene.Scene;
import javafx.stage.Stage;

public class Main extends Application {

    @Override
    public void start(Stage primaryStage) throws Exception{
        Parent root = FXMLLoader.load(getClass().getResource("sample.fxml"));
        primaryStage.setTitle("Hello World");
        primaryStage.setScene(new Scene(root, 300, 275));
        primaryStage.show();
    }


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

Controller :

package sample;

import javafx.beans.property.SimpleStringProperty;
import javafx.collections.FXCollections;
import javafx.collections.ObservableList;
import javafx.concurrent.Task;
import javafx.fxml.FXML;
import javafx.scene.control.TableColumn;
import javafx.scene.control.TableView;

public class Controller {
    public TableView<Integer> timeTable;
    public TableColumn<Integer, String> positionColumn;

    private ObservableList<Integer> testList;

    @FXML
    private void initialize() {
        testList = FXCollections.synchronizedObservableList(FXCollections.observableArrayList());
        positionColumn.setCellValueFactory(cellData -> new SimpleStringProperty(cellData.getValue().toString()));
        timeTable.setItems(testList);
        Task<Integer> integerTask = new Test(testList);
        Thread testThread = new Thread(integerTask);

        testThread.start();
    }
}

沟通任务:

package sample;

import javafx.collections.ObservableList;
import javafx.concurrent.Task;

public class Test extends Task<Integer> {
    private ObservableList<Integer> testlist;

    Test(ObservableList<Integer> list) {
        testlist = list;
    }

    @Override
    protected Integer call() throws Exception {
        // Emulates the server communication thread. Instead of an endless loop, I used a fixed number of iterations.
        // The real application has an endless while loop for server communication so a Task cannot be used to
        // get the data

        // getDataFromServer()
        // parseData()
        // putDataInList()
        // loop
        Thread.sleep(2000);
        for (int i = 0; i < 500; ++i) {
            testlist.add(i);
        }

        return 0;
    }
}

FXML:

<?import javafx.scene.control.*?>
<?import javafx.scene.layout.GridPane?>
<GridPane fx:controller="sample.Controller"
          xmlns:fx="http://javafx.com/fxml" alignment="center" hgap="10" vgap="10">
    <TableView fx:id="timeTable" editable="true" prefHeight="498.0"
               prefWidth="254.0">
        <columns>
            <TableColumn fx:id="positionColumn" prefWidth="73.0" text="Position"/>
        </columns>
        <columnResizePolicy>
            <TableView fx:constant="CONSTRAINED_RESIZE_POLICY"/>
        </columnResizePolicy>
    </TableView>
</GridPane>

Answer 1

即使列表是同步的，列表的修改仍然会触发执行修改的线程上的事件。在您的情况下，这会导致在非应用程序线程上触发 TableView 更新。

此外，您不能简单地使用 Task.updateValue，请参阅 the javadoc 的以下部分(强调我的):

Calls to updateValue are coalesced and run later on the FX application thread [...] and intermediate values may be coalesced to save on event notifications.

您需要自己同步。以下类结合更新来补偿快速更新，快速更新可能会通过发布许多可运行对象来减慢应用程序线程的速度:

public abstract class JavaFXWorker<T> implements Runnable {

    private final List<T> results = new LinkedList<>();
    private final Object lock = new Object();
    private boolean updateWaiting = false;

    protected void publish(T... values) {
        synchronized (lock) {
            for (T v : values) {
                results.add(v);
            }

            // don't trigger additional updates, if last update didn't fetch the results yet
            // this reduces the number of Runables posted on the application thread
            if (!updateWaiting) {
                updateWaiting = true;
                Platform.runLater(this::update);
            }
        }
    }

    private void update() {
        List<T> chunks;
        synchronized(lock) {
            // copy results to new list and clear results
            chunks = new ArrayList(results);
            results.clear();
            updateWaiting = false;
        }
        // run ui updates
        process(chunks);
    }

    protected abstract void process(List<T> chunks);

}

您的代码可以使用上面的类重写如下。

public class Test extends JavaFXWorker<Integer> {
    private final ObservableList<Integer> testlist;

    public Test(ObservableList<Integer> list) {
        testlist = list;
    }

    @Override
    public void run() {
        // Emulates the server communication thread. Instead of an endless loop, I used a fixed number of iterations.
        // The real application has an endless while loop for server communication so a Task cannot be used to
        // get the data

        // getDataFromServer()
        // parseData()
        // putDataInList()
        // loop
        Thread.sleep(2000);
        for (int i = 0; i < 500; ++i) {
            publish(i);
        }
    }

    @Override
    protected process(List<Integer> chunks) {
        testlist.addAll(chunks);
    }
}

testList = FXCollections.observableArrayList();

...

Thread testThread = new Thread(new Test(testList));

testThread.start();