多线程Client/Socket聊天应用多客户端连接卡顿问题求助

我正在开发一个多客户端单服务器的控制台聊天应用，已采用多线程实现，但仅单个客户端能正常运行，第二个客户端尝试连接时会出现卡顿。

应用概述 #

• 客户端输入WHOIS，服务器返回所有活跃客户端名称及加入时间；
• 输入普通消息，将消息发送至除发送者外的所有活跃客户端；
• 输入@client_name 消息，将消息发送给指定client_name的客户端；
• 输入LOGOUT，客户端退出；
• 输入PINGU，向所有活跃客户端发送企鹅相关趣味信息。

代码实现 #

Server代码 #

public class ChatServer {
    private static List<Socket> activeSockets = new ArrayList<>(50);

    public static List<CleverSocket> getCleverSockets() {
        return cleverSockets;
    }

    protected static List<Socket> getActiveSockets() {
        return activeSockets;
    }

    private static List<CleverSocket> cleverSockets = new ArrayList<>(50);

    public static void main(String[] args) throws IOException {
        int port = 3000;
        if (args.length > 0) {
            port = Integer.parseInt(args[0]);
        }
        ServerSocket server = null;
        try {
            server = new ServerSocket(port);
        } catch (IOException e) {
            server.close();
            throw new RuntimeException(e);
        }
        serverWaitingMsg(port);
        try (Socket socket = server.accept()) {
            var clientProcess = new NonBlockingClient(socket);
            Thread runClient = new Thread(clientProcess);
            runClient.start();

            while (true) {
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    protected static String connectionAccepted(Socket socket) {
        String hostName = socket.getInetAddress().getHostName();
        LocalTime now = LocalTime.now();
        String hostAddress = socket.getInetAddress().getHostAddress();
        String out = (now + " Connection accepted " + hostName + "/" + hostAddress);
        return out;
    }

    protected static void serverWaitingMsg(int port) {
        StringBuilder sb;
        sb = new StringBuilder();
        sb.append(LocalTime.now());
        sb.append(": Server is waiting on port/N ").append(port);
        System.out.println(sb);
    }

    protected static void clientJoinMsg(String clientInput) {
        StringBuilder sb;
        sb = new StringBuilder();
        LocalTime now = LocalTime.now();
        sb.append(now);
        sb.append(": *** ").append(clientInput).append(" has joined the chat room. ***");
        System.out.println(sb);
    }
}

class NonBlockingClient implements Runnable {
    private Socket client;

    public NonBlockingClient(Socket client) {
        this.client = client;
    }

    @Override
    public void run() {
        try (BufferedReader in = new BufferedReader(new InputStreamReader(client.getInputStream())); PrintWriter out = new PrintWriter(new OutputStreamWriter(client.getOutputStream()), true);) {
            //Read client's name
            String clientName = in.readLine();

            //Write that connection accepted!
            out.println(ChatServer.connectionAccepted(client));

            CleverSocket CLS = new CleverSocket(LocalTime.now(), clientName, client);

            ChatServer.getCleverSockets().add(CLS);
            ChatServer.getActiveSockets().add(client);

            //Print that client has joined
            ChatServer.clientJoinMsg(CLS.getName());

            //Now server is till waiting
            ChatServer.serverWaitingMsg(client.getPort());

            while (true) {
                String message = in.readLine();
                String result = parseMessage(message);
                if (result.equals("skip")) out.println("");
                else out.println(result);
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private String parseMessage(String message) {
        List<CleverSocket> cleverSockets = ChatServer.getCleverSockets();
        StringBuilder sb = new StringBuilder();
        sb.append(("List of all active clients: "));
        if (message.equalsIgnoreCase("whois")) {
            int i = 1;
            for (CleverSocket cs : cleverSockets) {
                sb.append("-" + i + ") " + cs.getName() + " since " + cs.getTime());
                i++;
            }

            return sb.toString();
        }

        if (message.contains("@")) {
            String filtered = message.substring(message.indexOf("@") + 1);
            String clientInfo[] = filtered.split(" ");
            ChatServer.getCleverSockets().forEach((cs) -> {
                if (cs.getName().equals(clientInfo[0])) {
                    String msg = clientInfo[1];
                    try {
                        var out = new PrintWriter(new OutputStreamWriter(cs.getSocket().getOutputStream()), true);
                        out.println(msg);
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                }
            });
            return "skip";
        }

        if (message.equalsIgnoreCase("pingu")) {
            for (CleverSocket cleverSocket : cleverSockets) {
                if (cleverSocket.getSocket() != client) {
                    try {
                        var out = new PrintWriter(new OutputStreamWriter(cleverSocket.getSocket().getOutputStream()), true);
                        out.println("Penguin is seabird!");
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                }
            }

            return "skip";
        }

        if (message.equalsIgnoreCase("logout")) {
            for (var it = ChatServer.getCleverSockets().iterator(); it.hasNext(); ) {
                CleverSocket tmp = it.next();
                if (tmp.getSocket() == client) {
                    it.remove();
                }
            }

            for (var it = ChatServer.getActiveSockets().iterator(); it.hasNext(); ) {
                var tmp = it.next();
                if (tmp == client) {
                    it.remove();
                }
            }

            return "skip";
        } else {
            ChatServer.getActiveSockets().forEach((as) -> {
                if (as != client) {
                    try {
                        var out = new PrintWriter(new OutputStreamWriter(as.getOutputStream()), true);
                        out.println(message);
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                }
            });

            return "skip";
        }
    }
}

class CleverSocket {
    private LocalTime time;
    private String name;
    private Socket socket;

    public LocalTime getTime() {
        return time;
    }

    public String getName() {
        return name;
    }

    public Socket getSocket() {
        return socket;
    }

    public CleverSocket(LocalTime time, String name, Socket socket) {
        this.time = time;
        this.name = name;
        this.socket = socket;
    }
}

Client代码 #

public class ChatClient {
    public static void main(String[] args) {
        String hostName = "localhost";
        int port = 3000;
        if (args.length > 0) {
            String tmp[] = args[0].split(":");
            hostName = tmp[0];
            port = Integer.parseInt(tmp[1]);
        }
        try (Socket client = new Socket(hostName, port);

             BufferedReader stdIn = new BufferedReader(new InputStreamReader(System.in)); BufferedReader in = new BufferedReader(new InputStreamReader(client.getInputStream())); PrintWriter out = new PrintWriter(new OutputStreamWriter(client.getOutputStream()), true);) {

            System.out.println("Enter your username: ");
            out.println(stdIn.readLine());
            System.out.println();
            //connection accepted
            System.out.println(in.readLine());

            welcome();

            String exit = "";
            while (!exit.equalsIgnoreCase("logout")) {
                String clientMsg = stdIn.readLine();
                exit = clientMsg;
                out.println(clientMsg);
                System.out.println();
                System.out.println(in.readLine());
            }
        } catch (UnknownHostException e) {
            throw new RuntimeException(e);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private static void welcome() {
        System.out.println();
        System.out.println("--- Welcome to the chatroom! ---");
        System.out.println("Instructions:");
        String text = """
                1. Simply type the message to broadcast it to all active clients!
                2. Type "@username<space>message" without quotes to send message to the specific client
                3. Type "WHOIS" without quotes to see list of active clients
                4. Type "LOGOUT" (not case sensitive) without quotes to logoff from server
                5. Type "PINGU" (not case sensitive) to req/send interesting fact about penguins
                """;
        System.out.println(text);
    }
}

问题原因 #

1. 服务器连接处理逻辑错误：服务器main方法仅调用一次server.accept()，处理完第一个客户端后进入死循环，再也不接受新的连接请求，导致后续客户端无法建立连接，出现卡顿。
2. 集合线程不安全：activeSockets和cleverSockets使用非线程安全的ArrayList，多线程环境下增删操作会引发并发异常或数据不一致。
3. 客户端消息读取阻塞：客户端主线程每次发送消息后调用in.readLine()等待响应，若其他客户端发来消息，会导致读取逻辑混乱，同时阻塞用户输入。

解决方案 #

1. 修复服务器连接监听逻辑 #

修改服务器main方法，将server.accept()放入循环，持续监听新客户端：

public static void main(String[] args) throws IOException {
    int port = 3000;
    if (args.length > 0) {
        port = Integer.parseInt(args[0]);
    }
    ServerSocket server = null;
    try {
        server = new ServerSocket(port);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
    serverWaitingMsg(port);
    // 循环接受客户端连接
    while (true) {
        try {
            Socket socket = server.accept();
            var clientProcess = new NonBlockingClient(socket);
            Thread runClient = new Thread(clientProcess);
            runClient.start();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

移除原代码中try(Socket socket = server.accept())块及内部死循环。

2. 替换为线程安全集合 #

将ArrayList替换为CopyOnWriteArrayList，避免并发操作问题：

private static List<Socket> activeSockets = new CopyOnWriteArrayList<>(50);
private static List<CleverSocket> cleverSockets = new CopyOnWriteArrayList<>(50);

3. 优化客户端消息读取逻辑 #

客户端单独开启线程监听服务器消息，避免主线程阻塞：

public static void main(String[] args) {
    String hostName = "localhost";
    int port = 3000;
    if (args.length > 0) {
        String tmp[] = args[0].split(":");
        hostName = tmp[0];
        port = Integer.parseInt(tmp[1]);
    }
    try (Socket client = new Socket(hostName, port);
         BufferedReader stdIn = new BufferedReader(new InputStreamReader(System.in));
         BufferedReader in = new BufferedReader(new InputStreamReader(client.getInputStream()));
         PrintWriter out = new PrintWriter(new OutputStreamWriter(client.getOutputStream()), true);) {

        System.out.println("Enter your username: ");
        out.println(stdIn.readLine());
        System.out.println();
        System.out.println(in.readLine());

        welcome();

        // 独立线程监听服务器消息
        new Thread(() -> {
            String serverMsg;
            try {
                while ((serverMsg = in.readLine()) != null) {
                    System.out.println("\n" + serverMsg);
                    System.out.print("Enter message: ");
                }
            } catch (IOException e) {
                System.out.println("Disconnected from server.");
            }
        }).start();

        String exit = "";
        while (!exit.equalsIgnoreCase("logout")) {
            System.out.print("Enter message: ");
            String clientMsg = stdIn.readLine();
            if (clientMsg == null) break;
            exit = clientMsg;
            out.println(clientMsg);
        }
    } catch (UnknownHostException e) {
        throw new RuntimeException(e);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}

同时移除服务器NonBlockingClient中if (result.equals("skip")) out.println("");代码，避免干扰客户端消息监听。

4. 复用输出流对象 #

修改CleverSocket类存储PrintWriter，避免重复创建流：

class CleverSocket {
    private LocalTime time;
    private String name;
    private Socket socket;
    private PrintWriter out;

    public LocalTime getTime() {
        return time;
    }

    public String getName() {
        return name;
    }

    public Socket getSocket() {
        return socket;
    }

    public PrintWriter getOut() {
        return out;
    }

    public CleverSocket(LocalTime time, String name, Socket socket, PrintWriter out) {
        this.time = time;
        this.name = name;
        this.socket = socket;
        this.out = out;
    }
}

在NonBlockingClient中初始化时传入PrintWriter：

PrintWriter out = new PrintWriter(new OutputStreamWriter(client.getOutputStream()), true);
CleverSocket CLS = new CleverSocket(LocalTime.now(), clientName, client, out);

发送消息时直接调用cs.getOut().println(msg)即可。

内容的提问来源于stack exchange，提问作者Giorgi Chapidze