java 8 - JavaFX 和 RxJava- TableView 无限调用 setCellValueFactory

Question

我在TableView上遇到了问题，并在setCellValueFactory中使用了ReactFX的反应式绑定。但是，驱动绑定的EventStream源自 RxJava Observable，在下面的代码中，您将找到一种将其转换为 EventStream 的方法。

但是，TableView从不显示列的初始绑定值以外的任何内容。当我向setCellValueFactory()体添加System.out.println时，我发现setCellValueFactory()在循环中被无限调用，并且发出的值从未进入绑定。

我对此感到非常困惑。如何停止此行为并让可观察对象成功将单个值发送到事件流，然后发出绑定？

这是我的SSCCE。

public class ReactiveTableViewTest extends Application {
    @Override
    public void start(Stage stage) throws Exception {
        Group root = new Group();
        Scene scene = new Scene(root);
        root.getChildren().add(new ReactiveTable(buildSampleData()));
        stage.setScene(scene);
        stage.show();
    }
    private ObservableList<ReactivePoint> buildSampleData() { 
        ObservableList<ReactivePoint> points = FXCollections.observableArrayList();
        points.add(new ReactivePoint(Observable.just(1), Observable.just(2)));
        return points;
    }
    private static final class ReactivePoint {
        private final Observable<Integer> x;
        private final Observable<Integer> y;
        ReactivePoint(Observable<Integer> x, Observable<Integer> y) { 
            this.x = x;
            this.y = y;
        }
        public Observable<Integer> getX() {
            return x;
        }
        public Observable<Integer> getY() { 
            return y;
        }
    }
    private static final class ReactiveTable extends TableView<ReactivePoint> {
        @SuppressWarnings("unchecked")
        private ReactiveTable(ObservableList<ReactivePoint> reactivePoints) { 
            this.setItems(reactivePoints);
            TableColumn<ReactivePoint,Number> xCol = new TableColumn<>("X");
            xCol.setCellValueFactory(cb -> {
                System.out.println("Calling cell value factory for x col");
                return toReactFX(cb.getValue().getX().map(x -> (Number) x)).toBinding(-1); //causes infinite call loop
                //return new SimpleObjectProperty<Number>(1); //works fine
            });
            TableColumn<ReactivePoint,Number> yCol = new TableColumn<>("Y");
            yCol.setCellValueFactory(cb -> {
                System.out.println("Calling cell value factory for y col");
                return toReactFX(cb.getValue().getY().map(y -> (Number) y)).toBinding(-1); //causes infinite call loop
                //return new SimpleObjectProperty<Number>(1); //works fine
            });
            this.getColumns().addAll(xCol, yCol);
        }
    }
    private static <T> EventStream<T> toReactFX(Observable<T> obs) {
        EventSource<T> es = new EventSource<>();
        obs.subscribe(foo -> Platform.runLater(() -> es.push(foo)), e -> e.printStackTrace());
        return es;
    }
    public static void main(String[] args) { 
        launch(args);
    }
}

更新

我想我在下面提出的解决方案发现了一个问题。如果在平台线程以外的其他线程上发出任何Observable，则不会向属性填充任何值。

我试图通过在将线程

放在平台线程上之前检查调用rxToProperty线程是否是平台线程来解决此问题，但这不起作用并再次导致无限循环。我不知道Property的线程安全是否正在破坏事情。

但是，如何在多个线程上发出可观察量以安全地填充Property呢？这是我更新的 SSCCE 显示此行为。"X"列永远不会填充，因为它是多线程的，但"Y"列会填充，因为它保留在平台线程上。

public class ReactiveTableViewTest extends Application {
    @Override
    public void start(Stage stage) throws Exception {
        Group root = new Group();
        Scene scene = new Scene(root);
        root.getChildren().add(new ReactiveTable(buildSampleData()));
        stage.setScene(scene);
        stage.show();
    }
    private ObservableList<ReactivePoint> buildSampleData() { 
        ObservableList<ReactivePoint> points = FXCollections.observableArrayList();
        points.add(new ReactivePoint(
                Observable.just(1, 5, 6, 8,2,3,5,2).observeOn(Schedulers.computation()), 
                Observable.just(2,6,8,2,14)
                )
            );
        return points;
    }
    private static final class ReactivePoint {
        private final Observable<Integer> x;
        private final Observable<Integer> y;
        ReactivePoint(Observable<Integer> x, Observable<Integer> y) { 
            this.x = x;
            this.y = y;
        }
        public Observable<Integer> getX() {
            return x;
        }
        public Observable<Integer> getY() { 
            return y;
        }
    }
    private static final class ReactiveTable extends TableView<ReactivePoint> {
        @SuppressWarnings("unchecked")
        private ReactiveTable(ObservableList<ReactivePoint> reactivePoints) { 
            this.setItems(reactivePoints);

            System.out.println("Constructor is happening on FX THREAD: " + Platform.isFxApplicationThread());
            TableColumn<ReactivePoint,Number> xCol = new TableColumn<>("X");
            xCol.setCellValueFactory(cb -> { 
                System.out.println("CellValueFactory for X called on FX THREAD: " + Platform.isFxApplicationThread());
                return rxToProperty(cb.getValue().getX().map(x -> (Number) x));
                }
            );
            TableColumn<ReactivePoint,Number> yCol = new TableColumn<>("Y");
            yCol.setCellValueFactory(cb -> {
                System.out.println("CellValueFactory for Y called on FX THREAD: " + Platform.isFxApplicationThread());
                return rxToProperty(cb.getValue().getY().map(y -> (Number) y));
            }
        );
            this.getColumns().addAll(xCol, yCol);
        }
    }
    private static <T> ObjectProperty<T> rxToProperty(Observable<T> obs) { 
        ObjectProperty<T> property = new SimpleObjectProperty<>();
        obs.subscribe(v -> { 
            if (Platform.isFxApplicationThread()) {
                System.out.println("Emitting " + v + " on FX Thread");
                property.set(v);
            }
            else { 
                System.out.println("Emitting " + v + " on Non-FX Thread");
                Platform.runLater(() -> property.set(v));
            }
        });
        return property;
    }
    public static void main(String[] args) { 
        launch(args);
    }
}

Answer 1

我找到了一个解决方案，尽管我还没有完全弄清楚OP代码中问题的确切根本原因（如果有人可以启发原因，我会将其标记为答案）。我最初认为Platform.runLater()可能会导致最初设置的无限循环（如下所示）。

private static <T> EventStream<T> toReactFX(Observable<T> obs) {
    EventSource<T> es = new EventSource<>();
    obs.subscribe(foo -> Platform.runLater(() -> es.push(foo)), e -> e.printStackTrace());
    return es;
}

事实证明，这个理论是正确的。移除Platform.runLater()导致无限循环消失。也许发出的值一直扔到 GUI 线程的后面，因此从未进入绑定？但仍然没有发出任何内容，表值保持在 -1，即初始绑定值。

private static <T> EventStream<T> toReactFX(Observable<T> obs) {
    EventSource<T> es = new EventSource<>();
    obs.subscribe(foo -> es.push(foo), e -> e.printStackTrace());
    return es;
}

我确实找到了一些有用的东西。我创建了一个名为 rxToProperty() 的新转换方法，并用它替换了对rxToReactFX()的调用。在那之后，一切似乎都很好。

private static <T> ObjectProperty<T> rxToProperty(Observable<T> obs) { 
    ObjectProperty<T> property = new SimpleObjectProperty<>();
    obs.subscribe(v -> property.set(v));
    return property;
}

以下是新的TableColumn设置。

TableColumn<ReactivePoint,Number> xCol = new TableColumn<>("X");
xCol.setCellValueFactory(cb -> rxToProperty(cb.getValue().getX().map(x -> (Number) x)));
TableColumn<ReactivePoint,Number> yCol = new TableColumn<>("Y");
yCol.setCellValueFactory(cb -> rxToProperty(cb.getValue().getY().map(y -> (Number) y)));

如果有人有更好的解决方案，或者可以解释为什么Platform.runLater()和EventStream不起作用，我会将其标记为可接受的答案。

更新

在

非 FX 线程上发出的可观察量存在一些问题，并且值从未填充到Property。我发现这可以通过使用 cache() 来保存最后一个值来解决，它将在订阅的调用线程（即 FX 线程）上重新=emit。我还做了一些同步并只读返回Property的包装器。

private static <T> ReadOnlyObjectProperty<T> rxToProperty(Observable<T> obs) { 
        ReadOnlyObjectWrapper<T> property = new ReadOnlyObjectWrapper<>();
        obs.cache(1).subscribe(v -> { 
            synchronized(property) { 
                if (Platform.isFxApplicationThread()) {
                    System.out.println("Emitting val " + v + " on FX Thread");
                }
                else { 
                    System.out.println("Emitting val " + v + " on Non-FX Thread");
                }
                property.set(v);
            }
        });
        return property.getReadOnlyProperty();
    }

最后更新Tomas Mikula 在 ReactFX GitHub 项目上对这种行为以及解决方案给出了一些非常有用的见解。

https://github.com/TomasMikula/ReactFX/issues/22