Java并发网络问题

Question

我是一名试图学习网络编程和并发性的 Java 新手，我想尝试编写一个简单的聊天服务器，其中来自客户端的输入会回显给所有客户端。那并没有发生。我添加了几个打印语句，以便程序在每次收到连接时都会宣布它正在等待连接。我在本地使用 Telnet 连接到我的计算机上的端口。

程序宣布第一个和第二个并发连接成功，但随后不会宣布后续连接成功直到我关闭所有连接。因此，例如，我将从五个单独的终端进行连接，并且程序将宣布“连接 1”和“连接 2”，但不会宣布“连接 3”、4 和 5，直到我关闭所有终端。

我正在寻求帮助找出我的错误所在，以及如何调试此类情况的一般建议。

简而言之，我的程序有

1. 主类，启动其他三个线程
2. 一个 ClientListener 类，它使用 SocketReader 来监听连接并将 Sockets 输入流和输出流存储在两个 Set 中。
3. 一个 MessageReader，它迭代输入流。如果它找到消息，则会将其放入 SynchronousQueue 中并等待
4. MessageWriter 将其删除。 MessageWriter 将消息发送到所有输出流。

代码如下。感谢您的帮助!

public class Main {

    public static void main(String[] args) {
        ClientListener clientListener = new ClientListener();
        Thread clientListenerThread = new Thread(clientListener);
        clientListenerThread.setPriority(Thread.MAX_PRIORITY);
        clientListenerThread.start();

        MessageReader messageReader = new MessageReader(clientListener);
        Thread messageReaderThread = new Thread(messageReader);
        messageReaderThread.setPriority(Thread.MIN_PRIORITY);
        messageReaderThread.start();

        MessageWriter messageWriter = new MessageWriter(messageReader, clientListener);
        Thread messageWriterThread = new Thread(messageWriter);
        messageWriterThread.setPriority(Thread.NORM_PRIORITY);
        messageWriterThread.start();
    }
}

public class ClientListener implements Runnable {
    private static final int DEFAULT_PORT = 5000;

    private Set<Scanner> clientIn = Collections.synchronizedSet(
            new LinkedHashSet<Scanner>());
    private Set<PrintWriter> clientOut = Collections.synchronizedSet(
            new LinkedHashSet<PrintWriter>());

    public Set<Scanner> getClientIn() {
        return clientIn;
    }

    public Set<PrintWriter> getClientOut() {
        return clientOut;
    }

    @Override
    public void run() {
        try {
            ServerSocket server = new ServerSocket(DEFAULT_PORT);
            System.out.println("Listening for connections...");
            int connectionNum = 0;

            while(true) {
                Socket socket = server.accept();
                connectionNum++;
                System.out.format("Connection %s%n", connectionNum);

                Scanner in = new Scanner(socket.getInputStream());
                PrintWriter out = new PrintWriter(socket.getOutputStream());
                clientIn.add(in);
                clientOut.add(out);
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}


public class MessageReader implements Runnable {
    private ClientListener clientListener;
    private BlockingQueue<String> messages = new SynchronousQueue<String>();

    public MessageReader(ClientListener clientListener) {
        this.clientListener = clientListener;
    }

    @Override
    public void run() {
        while(true) {
            Set<Scanner> clients = clientListener.getClientIn();
            synchronized (clients) {
                for(Scanner client: clients) {
                    if(client.hasNext()) {
                        try {
                            messages.put(client.next());
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }
        }
    }

    public String getMessage() throws InterruptedException {
        return messages.take();
    }
}


public class MessageWriter implements Runnable {
    private ClientListener clientListener;
    private MessageReader messageReader;

    public MessageWriter(
            MessageReader messageReader, 
            ClientListener clientListener) {
        this.messageReader = messageReader;
        this.clientListener = clientListener;
    }

    @Override
    public void run() {
        try {

            while(true) {
                String message = messageReader.getMessage();

                Set<PrintWriter> clients = clientListener.getClientOut();
                synchronized (clients) {
                    for(PrintWriter client: clients) {
                        client.println(message);
                    }
                }
            }

        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Answer 1

我不是线程专家，但是在 MessageReader 类中有这一行

if(client.hasNext())

Scanner.hasNext() 的 Javadoc 说“此方法在等待扫描输入时可能会阻塞。扫描仪不会前进超过任何输入。”

如果扫描器仍处于等待状态，则同步方法将永远不会继续并阻止所有其他输入。正如我之前的评论中所说，类 ClientListener 中的 clientIn.add(in); 行可能会被阻止，因为它是一个同步 Set，但由于 print 语句写在它之前，可能会给人这样的印象:连接 2 已成功建立。