java - 如何从JavaFX中的二级线程获取更新消息/属性？

Question

在 JavaFX 应用程序中，我有一个主线程，它通过套接字监听多个客户端，并为每个客户端连接创建单独的线程用于与客户端通信。我想显示来自 JavaFX 场景中所有客户端的消息(通过属性绑定(bind)或其他方式)。我尝试使用任务和服务，如 JavaFX 并发教程 https://docs.oracle.com/javase/8/javafx/interoperability-tutorial/concurrency.htm 中所述。 。我可以通过 updateMessage 从主线程获取消息，但无法弄清楚如何从为客户端通信创建的第二级线程获取消息。下面给出了我尝试的示例代码。

public class SocketService
    {
        int portNumber;

        public SocketService(int portNumber)
        {
            this.portNumber = portNumber;

            Task task = new Task<Void> ()
            {
                @Override
                public Void call()
                {
                    try (ServerSocket serverSocket = new ServerSocket(portNumber))
                    {
                        while (true)
                        {
                            ClientService cs = new ClientService(serverSocket.accept());
                            cs.start();
                            updateMessage("Connected");
                        }
                    } catch (IOException e)
                    {
                        System.out.println("Exception!!! "+e.getMessage());
                    }
                    return null;
                }
            };
           new Thread(task).start();
        }
    }

    class ClientService extends Service<Void>
    {
        Socket connectSocket;
        public ClientService(Socket connectSocket)
        {
            this.connectSocket = connectSocket;
        }

        @Override
        public Task<Void> createTask()
        {
            return new Task<Void>()
            {
                @Override public Void call()
                {
                    try (PrintWriter out = new PrintWriter(connectSocket.getOutputStream(), true);
                         BufferedReader in = new BufferedReader(new InputStreamReader(connectSocket.getInputStream()));)
                    {
                        String inText;

                        while ((inText = in.readLine()) != null)
                        {
                            System.out.println(inText);
                            updateMessage(inText); // want to get this message updated in JavaFX
                        }
                    } catch (IOException e)
                    {
                        System.out.println("Exception!!! "+e.getMessage());
                    }
                    return null;
                }
            };
        }
    }

预先感谢您的帮助。

Answer 1

您可以观察您创建的服务的消息属性。该属性更改的通知在 FX 应用程序线程上执行(请注意，顺便说一句，您有一个错误:您只能从 FX 应用程序线程调用 Service.start()；即已修复):

while (true) {
    ClientService cs = new ClientService(serverSocket.accept());
    cs.messageProperty().addListener((obs, oldMessage, newMessage) -> {
        // update UI with newMessage...
    }
    Platform.runLater(cs::start);
    updateMessage("Connected");
}

对于更复杂的方法，您可能不希望向 SocketService 类公开 UI 的详细信息。所以你可能会考虑类似的事情

public class SocketService {

    private final ObservableMap<ClientService, String> messageLookup 
        = FXCollections.observableMap(new HashMap<>()) ;

    private final ObservableList<String> messages = FXCollections.observableArrayList();

    public ObservableList<String> getMessages() { return messages ; }

    int portNumber ;

    public SocketService(int portNumber) {

        this.portNumber = portNumber ;

        messageLookup.addListener((Change<? extends ClientService, ? extends String> change) -> {
            if (change.wasAdded()) {
                messages.add(change.getValueAdded());
            }
            if (change.wasRemoved()) {
                messages.remove(change.getValueRemoved());
            }
        });

        Task<Void> task = new Task<Void>() {
            @Override
            public void call() {
                try (ServerSocket serverSocket = new ServerSocket(portNumber)) {
                    while(true) {
                        ClientService cs = new ClientService(serverSocket.accept());
                        cs.messageProperty().addListener((obs, oldMessage, newMessage) -> 
                            messageLookup.put(cs, newMessage));
                        Platform.runLater(cs::start);
                        updateMessage("Connected");
                    }
                } catch (...) { ... }
                return null ;
            }
        }
    }
}

现在在您的用户界面中您可以执行以下操作

ListView<String> messages = new ListView<>();
SocketService socketService = ... ;
messages.setItems(socketService.getMessages());

显然，您如何使用它取决于您正在做什么以及您想要如何显示客户端的状态，但这应该给您一个想法。

这是一个完整的、可运行的示例:

import java.util.HashMap;
import java.util.Random;

import javafx.application.Application;
import javafx.application.Platform;
import javafx.collections.FXCollections;
import javafx.collections.MapChangeListener.Change;
import javafx.collections.ObservableList;
import javafx.collections.ObservableMap;
import javafx.collections.transformation.SortedList;
import javafx.concurrent.Service;
import javafx.concurrent.Task;
import javafx.geometry.Pos;
import javafx.scene.Scene;
import javafx.scene.control.Label;
import javafx.scene.control.ListView;
import javafx.scene.layout.VBox;
import javafx.stage.Stage;

public class TaskMessageUpdateExample extends Application {



    @Override
    public void start(Stage primaryStage) {

        ObservableMap<ClientService, String> messageLookup = FXCollections.observableMap(new HashMap<>());
        ObservableList<String> messages = FXCollections.observableArrayList();

        messageLookup.addListener((Change<? extends ClientService, ? extends String> change) -> {
            if (change.wasAdded()) {
                messages.add(change.getValueAdded());
            }
            if (change.wasRemoved()) {
                messages.remove(change.getValueRemoved());
            }
        });

        Task<Void> serverTask = new Task<Void>() {

            private Random rng = new Random();

            @Override
            public Void call() throws Exception {
                int nClients = rng.nextInt(11);
                for (int i = 1 ; i <= nClients; i++) {
                    Thread.sleep(rng.nextInt(2000)+500);
                    ClientService cs = new ClientService();
                    String clientID = "Client "+i ;
                    updateMessage("Connected "+clientID);
                    cs.messageProperty().addListener((obs, oldMessage, newMessage) -> {
                        messageLookup.put(cs, clientID +": " +newMessage);
                    });
                    Platform.runLater(cs::start);
                }
                return null ;
            }
        };

        Thread t = new Thread(serverTask);
        t.setDaemon(true);
        t.start();

        Label statusLabel =  new Label();
        statusLabel.textProperty().bind(serverTask.messageProperty());

        ListView<String> listView = new ListView<>();
        listView.setItems(new SortedList<String>(messages, String::compareTo));

        VBox root = new VBox(5, listView, statusLabel);
        root.setAlignment(Pos.TOP_CENTER);
        Scene scene = new Scene(root, 250, 600);
        primaryStage.setScene(scene);
        primaryStage.show();
    }

    private static class ClientService extends Service<Void> {

        @Override
        protected Task<Void> createTask() {
            Task<Void> task =  new Task<Void>() {
                @Override
                public Void call() throws InterruptedException {
                   Random rng = new Random();
                   for (int i = 1 ; i <= 10 ; i++) {
                       Thread.sleep((rng.nextInt(1000)+500));
                       updateMessage("Step "+i);
                   }
                   return null ;
                }
            };
            return task ;
        }

    }

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