Java并发网络问题

Question

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

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

我正在寻找帮助，找出我的错误在哪里以及如何处理这种情况的一般建议。

概括地说，我的程序有

1. 主类，这将启动其他三个线程
2. 一个ClientListener类，它采用了SocketReader侦听连接和存储插槽inputstreams和outputstreams在两套。
3. 一个MessageReader，它遍历输入流。如果它找到一条消息，它将它放入一个SynchronousQueue中，并等待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()) 

的Javadoc Scanner.hasNext（）说的“这种方法可能在等待要扫描的输入块。扫描仪不会超过任何输入。“

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