基于套接字的应用程序的非确定性行为

Question

我正在开发客户端-服务器应用程序，但遇到了一个问题。我得到了负责一个连接客户端的ServerWorker，它创建了 2 个线程，1 个用于侦听来自该客户端的传入数据，1 个用于向他发送数据。

class ServerWorker {
    private DataProcessor dataProcessor;
    private ObjectInputStream inputStream;
    private ObjectOutputStream outputStream;
    private Thread receiverThread;
    private Thread senderThread;
    private Optional<DataPacket> dataToSend;

    private ServerWorker(Socket socket) {
        try {
            dataToSend = Optional.empty();
            dataProcessor = new DataProcessor();
            receiverThread = new Thread(this::readAndProcessData);
            senderThread = new Thread(this::sendData);
            inputStream = new ObjectInputStream(socket.getInputStream());
            outputStream = new ObjectOutputStream(socket.getOutputStream());
        } catch (IOException e) {
            //TODO
        }
    }

    static ServerWorker create(Socket socket) {
        return new ServerWorker(socket);
    }

    void start() {
        receiverThread.start();
        senderThread.start();
    }

    void stop() {
        receiverThread.interrupt();
        senderThread.interrupt();
    }

    private void readAndProcessData() {
        DataPacket dataPacket;
        try {
            while((dataPacket = (DataPacket)inputStream.readObject()) != null) {
                System.out.println("incoming message: " + dataPacket.getContent());
                dataToSend = Optional.of(dataProcessor.process(dataPacket));
            }
        } catch (ClassNotFoundException | IOException e) {
            //TODO
        }
    }

    private void sendData() {
        while(true) { //TODO
            dataToSend.ifPresent(data -> {
                try {
                    outputStream.writeObject(data);
                    outputStream.flush();
                    dataToSend = Optional.empty();
                } catch (IOException e) {
                    //TODO
                }
            });
        }
    }
}

而DataProcessor目前只是一个小班

public class DataProcessor {
   public DataPacket process(DataPacket packet){
        packet.setContent(packet.getContent().toUpperCase());
        return packet;
   }
}

当然，DataPacket 对客户端和服务器都是一样的

public class DataPacket implements Serializable {
    private String content;

    public String getContent() {
        return content;
    }

    public void setContent(String content) {
        this.content = content;
    }
}

DataProcessor 和 DataPackets 现在只是一种 POC，这将成长为更大更复杂的类，长话短说，ServerWorker 将接收数据并将其传递给处理，然后在完成一些逻辑之后，返回的数据将存储在dataToSend变量中，发送后删除。问题是，我在上面发布的代码仅在某些时候有效。当我运行我的服务器应用程序和客户端一（下面的代码）时，90% 的时间都没有发生任何事情，大写的“hello world”不会返回到客户端。有趣的是，当我在调试模式下运行我的服务器时（即使没有任何断点！），它也可以工作......知道到底出了什么问题吗？

 public static void main(String... args) throws Exception {
        Socket socket = new Socket("localhost", 9999);
        DataPacket dataPacket = new DataPacket();
        dataPacket.setContent("hello world");
        ObjectOutputStream os = new ObjectOutputStream(socket.getOutputStream());
        os.writeObject(dataPacket);
        os.flush();

        ObjectInputStream inputStream = new ObjectInputStream(socket.getInputStream());
        while((dataPacket = (DataPacket)inputStream.readObject()) != null) {
            System.out.println(dataPacket.getContent());
        }
    }

编辑# 添加一个类，ConnectionDispatcher，负责创建 ServerWorker 对象

class ConnectionDispatcher implements Runnable {
    private ServerSocket serverSocket;
    private List<ServerWorker> serverWorkers;
    private volatile boolean isReceiving;

    private ConnectionDispatcher(int port) throws IOException {
        serverSocket = new ServerSocket(port);
        serverWorkers = new ArrayList<>();
        isReceiving = false;
    }

    static ConnectionDispatcher create(int port) throws IOException {
       return new ConnectionDispatcher(port);
    }

    @Override
    public void run() {
        isReceiving = true;
        while(isReceiving) {
           acceptIncomingConnections();
        }
    }

    private void acceptIncomingConnections() {
        try {
            ServerWorker worker = ServerWorker.create(serverSocket.accept());
            serverWorkers.add(worker);
            worker.start();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

Answer 1

您似乎正在从一个线程修改dataToSend，同时在另一个线程中读取它的值。这不是线程安全的，正在读取其值的线程可能永远不会看到其他线程设置的更新值。因此，我将dataToSend 声明为volatile。

private volatile Optional<DataPacket> dataToSend;

我还没有机会亲自测试一下，但我可以在大约一个小时内进行测试（假设此更改不能解决您的问题）。

Answer 2

您可以使用数组阻塞队列来模拟生产者和消费者模式。

让接收者线程，将新的DataPacket放入队列，让发送者从队列中取出并处理后发送。

这将消除线程问题并充当缓冲区。

使用您当前的代码，当数据包以更高的速率到达时，您可能会丢失数据包。

我同意 user930，private Optional<DataPacket> dataToSend; 应该是易变的。

您还可以使用 JavaNIO 使您的代码具有更高的可扩展性，您可以查看 Apache Mina 项目。