Java 是等待和从客户端获取数据的最佳方式

Question

我开始使用 JDK 中的主要网络包学习网络，通过几个示例后它非常简单易行。但现在我对制作像聊天系统这样的多客户端应用程序很感兴趣。

目前我的结构思路是这样的：

连接处理程序类，它处理传入的连接，并保存客户端列表。 如果找到新连接，则创建一个新的客户端对象，启动它的线程（客户端对象将实现可运行，因此它会启动它自己的循环服务，它会循环接收到的新数据包），并将其添加到列表中。

我为每个客户端创建一个新线程，而不是循环遍历所有客户端，因为从客户端进程读取会停止整个执行并等待客户端发送数据，这让我有点恼火，这是我的问题。

我创建了一个从客户端接收消息的简单控制台应用程序，但现在我想检测断开连接。我读到，如果用户未连接，则 bufferedReader .read() 方法返回 -1，所以我认为我可以循环并每隔几秒对每个客户端执行一次，但问题是，客户端必须发送一个数据包才能 . read() 它，所以假设你这样做 .read() 它将等待并停止整个线程，直到收到数据包，（我认为）。

这是我当前从客户端获取消息的代码：

public boolean isConnected() {
    try {
        this.in.read();
        this.lastCheck = System.currentTimeMillis();
        return true;
    } catch (IOException e) {
        if (!inConnection()) {
            System.out.println("User disconnected");
            try {
                this.destruct();
            } catch (IOException e1) {
                e1.printStackTrace();
            }
        }
    }
    return false;
}

private boolean inConnection() {
    return System.currentTimeMillis() - lastCheck < this.maxTime;
}

public void startClientService() throws IOException {
    while(!this.session.isClosed()) {
        if (System.currentTimeMillis() - this.checkTime > 600) {
            System.out.println(System.currentTimeMillis() - this.checkTime);
            if (this.isConnected()) {
                int packetType = this.dataIn.readInt();
                packets.getPacket(packetType);
            }
        }
    }
}

public void destruct() throws IOException {
    this.session.close();
    this.connection.removeClient(this);
    System.out.println("Session killed");
}

基本上这里发生了什么，我从客户端发送一个打包的整数，我可能有很多事情要做，因此我可以设置许多唯一的数据包 ID，所以如果我想接收和处理聊天消息，数据包 id是216，客户端发送一个int 216，服务器读取数据包，进入所有数据包ID的交换循环并检测它是否真的是216，如果是，它获取处理消息的打包类的实例并获取接收到的字节像这样的消息：

public class Chat implements Packet {

    @Override
    public void processPacket(Session c) {
        String message = readMessage(c);
        System.out.println("Message: " + message);
    }

    private String readMessage(Session c) {
        byte[] data = c.readBytes();
        String message = null;
        try {
            message = new String(data, "UTF-8");
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        return message;
    }
}

这就是我读取字节的方式：

public byte[] readBytes() {
    int len;
    byte[] data = null;
    try {
        len = this.dataIn.readInt();
        data = new byte[len];
        if (len > 0) {
            this.dataIn.readFully(data);
        }
    } catch (IOException e) {
        e.printStackTrace();
    }       
    return data;
}

好的，我的问题：

添加断开连接检测后，当我发送消息时，没有任何反应。这可能是由于 .read() 它停止并等待响应。但是如果我再写一条消息，我会在服务器中收到消息。

这是我临时的、丑陋的客户：

public class Client {

    public static void main(String[] args) throws UnknownHostException, IOException {
        Socket socket = new Socket("127.0.0.1", 43594);
        Scanner r = new Scanner(System.in);
        PrintWriter out = new PrintWriter(socket.getOutputStream());
        String input;
        while(true) {
            input = r.next();
            if (input != null) {
                sendMessage(input, out);
            }
        }

    }

    public static void sendMessage(String message, PrintWriter out) {
        byte[] encoded = encode(message);
        out.write(0);
        out.println(encoded + "\n");
        out.flush();
    }

    public static byte[] encode(String s) {
        return DatatypeConverter.parseBase64Binary(s);
    }

    public static String decode(byte[] s) {
        return DatatypeConverter.printBase64Binary(s);
    }   
}

我的问题是：从客户端读取数据而不让应用程序等待它并实际每次循环的更好方法是什么？或者也许我应该有一个新线程来检查用户是否在线，所以每个客户端有 2 个线程？

如果有人需要我的会话对象（客户端对象）：

public class Session extends Thread implements Runnable {

    private Socket session;
    private Client client;
    private PrintWriter out;
    private BufferedReader in;  
    private PacketHandler packets;
    private DataInputStream dataIn;
    private ConnectionHandler connection;

    private final int checkTime = 1600;
    private final int maxTime = 22000;
    private long lastCheck;

    public Session(Socket session) {
        this.session = session;
        this.client = new Client(this);
        try {
            this.setStream();
        } catch (IOException e) {
            e.printStackTrace();
        }
        this.packets = new PacketHandler(this);
        System.out.println("[New session created]: " + session.getRemoteSocketAddress());
    }   

    public void setConnectionHandler(ConnectionHandler c) {
        this.connection = c;
    }

    public void run() {
        try {
            this.startClientService();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void setStream() throws IOException {
        this.out = new PrintWriter(this.session.getOutputStream());
        this.in = new BufferedReader(new InputStreamReader(this.session.getInputStream()));
        this.dataIn = new DataInputStream(this.session.getInputStream());
    }

    public Client getClient() {
        return this.client;
    }

    public byte[] readBytes() {
        int len;
        byte[] data = null;
        try {
            len = this.dataIn.readInt();
            data = new byte[len];
            if (len > 0) {
                this.dataIn.readFully(data);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }       
        return data;
    }

    public String readMessage() {
        try {
            return this.in.readLine();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    public boolean isConnected() {
        try {
            this.in.read();
            this.lastCheck = System.currentTimeMillis();
            return true;
        } catch (IOException e) {
            if (!inConnection()) {
                System.out.println("User disconnected");
                try {
                    this.destruct();
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
            }
        }
        return false;
    }

    private boolean inConnection() {
        return System.currentTimeMillis() - lastCheck < this.maxTime;
    }

    public void startClientService() throws IOException {
        while(!this.session.isClosed()) {
            if (System.currentTimeMillis() - this.checkTime > 600) {
                System.out.println(System.currentTimeMillis() - this.checkTime);
                if (this.isConnected()) {
                    int packetType = this.dataIn.readInt();
                    packets.getPacket(packetType);
                }
            }
        }
    }

    public void destruct() throws IOException {
        this.session.close();
        this.connection.removeClient(this);
        System.out.println("Session killed");
    }

}

谢谢！

Answer 1

虽然我没有时间查看所有代码，但这里有两件事可以帮助您。

1) 使用定义的消息头。定义客户端将发送到服务器的每条消息的 X 字节数。使用这些字节来定义消息的长度以及消息的类型。服务器知道此标头的长度和布局，并使用它以特定方式处理消息。示例可以是一个字节的标头。值 1 可能是“我已连接”消息。 2可能是我即将断开连接。 3 可能是我目前不在，而 4 可能是收到的聊天消息。

2) 有两种方法可以处理输入。首先是使用阻塞IO，并创建一个单独的线程来接收来自每个客户端的消息。我相信这就是你目前正在做的事情。第二种是使用非阻塞 IO，并有一个单独的线程遍历打开的套接字并进行读取。非阻塞会检查是否有数据要读取，如果没有，线程会继续执行。