java - tcp 服务器 : why adding delay increases ability to service clients? 的异常行为

Question

大家好!

我的问题是关于基于 NIO 的服务器，我的情况如下:服务器从 100 个客户端(100 个客户端线程)读取消息，每个客户端发送 100 条消息。因此，消息总数为 100x100 = 10000。我的服务器中有传入消息计数器，它在从某个客户端读取消息后增加。当我刚刚阅读消息时，我的服务器读取了大约 9200 条消息。 当我为模拟服务延迟添加虚拟循环时，我的服务器出人意料地为所有 10000 条消息提供服务!

我的期望是这样的——好吧，即使有很短的延迟，服务器也会设法读取所有 10000 条消息。因此，如果没有这种延迟，服务器可能可以读取更多消息(服务更多客户端)。但是你看，这是错误的。事不宜迟，事情会变得更糟。 Here我描述了我的架构。我当前逻辑的唯一修改是将接受客户端和读取消息分为 2 个不同的线程:一个选择器在一个线程中接受客户端，第二个选择器在另一个线程中等待来自已连接客户端的消息。

客户端代码

public class TCPClient implements Runnable{

private String name;
private static TCPClient[] clients;
private static Thread[] threads;
private int counter = 0;

public TCPClient(String name)
{
  this.name = name;
  this.counter = 0;
}

public static void main(String[] args) throws Exception
{
   clients = new TCPClient[100];
   threads = new Thread[100];
   for(int i=0;i<100;i++) 
   {
       clients[i] = new TCPClient("thread # "+Integer.toString(i));
       threads[i] = new Thread(clients[i]);
       threads[i].start(); 
       // Thread.sleep(500);
   }
   for(int i=0;i<100;i++) 
   {
       threads[i].join();
   }
   for(int i=0;i<100;i++){
       System.out.println("counter = "+clients[i].counter);
   }

}

@Override
public void run()
{
   Socket socket = null;
   OutputStream out = null;

   try
   {
        socket = new Socket();
        socket.connect(new InetSocketAddress("192.168.3.109",2345), 0); 
        out =  socket.getOutputStream();

        byte[] bytes;
        while(counter < 100)
        {
            counter++;
            bytes = (name+ ", message # "+Integer.toString(counter)+System.lineSeparator()).getBytes();
           // System.out.println(counter);
            out.write(bytes); 
            out.flush();
            Thread.sleep(200);
        }
   }
   catch(Exception ex)
   {
            System.out.println(name+" "+Integer.toString(counter));
            ex.printStackTrace(new PrintStream(System.out));
            System.out.println();
   }
    finally
    { 
        if(socket!=null && out!=null)
        {
            try
            {
                socket.close();
                out.close();
            }
            catch(Exception ex)
            {
                System.out.println("client close error");
            }
        }
    }
}

}

服务端代码(消息读取部分)

@Override
public void run() 
{
    isRunning = true;
    int acc = 0;
    boolean error = false;
    while (isRunning) {
      try 
      {


             selector.select(); 

             Set keys =  selector.selectedKeys();
             Iterator it = keys.iterator(); 
             while(it.hasNext())
             {
                 SelectionKey key = (SelectionKey)it.next();



                 if (key.isReadable())
                 { 
                       //readMessage(key);
                       //key.cancel();

                      // ByteBuffer bbb =  ByteBuffer.allocate(2048);
                      // key.cancel();
                       curTime = System.currentTimeMillis();
                       SocketChannel sc = (SocketChannel) key.channel();
                       // System.out.println("before reading"); 
                       bb.clear();
                       int x =  sc.read(bb); 

                       if(x==-1)
                       {
                           key.cancel();
                           //System.out.println("cancelling key");
                           continue;
                       }

                       counter++;
                      // bb.flip();
                       //System.out.print(decoder.decode(bb).toString());
                     //  Thread.sleep(20);
                       long sum=0;
                       for(int dummy=0;dummy<250000;dummy++)
                       {
                           sum += dummy;
                         //  sum %= 1005;
                       }
                       long delta  = System.currentTimeMillis() - curTime;
                       serviceTime += delta;

                       if(counter>9000)
                       {
                           System.out.println("recieved messages count = "+counter);
                           System.out.println("one message  service time = "+delta+" milliseconds");
                           System.out.println("total service time = "+serviceTime+" milliseconds");
                           System.out.println("sum = "+sum); //11 249 925 000
                       }

                     // selector.wakeup();
                      //key.interestOps(SelectionKey.OP_READ); 


                 } 
             }

             keys.clear();





      } 
      catch (Exception ex) 
      {     
          error = true;  
          System.out.println("error in recieving messages "+ex.getMessage());
          ex.printStackTrace(new PrintStream(System.out));
         // logger.println("error in recieving messages "+ex.getMessage());
         // logger.flush();
      }
      finally
      {
            //if(error) // !!!!!!!!!!! DO NOT STOP THE SERVER EDIT IT LATER
            //stopServer();
      }
    }
}

可能有用的信息 - 客户端每 2 条消息之间的延迟为 200 毫秒。当虚拟循环使 200000-220000 次迭代服务器完美运行时。顺便说一句，200000 次迭代大约是 200 毫秒——因为客户端数量是 100，所以一个 select() 的延迟是 100*200000 = 200 万次迭代——对于现代 PC 来说是 200 毫秒。如果虚拟循环的迭代次数少于 200000 次，服务器将读取 ~9200 条消息。这种奇怪行为的原因是什么？

Answer 1

这里有太多问题，很难知道从哪里开始。

1. 当 read() 返回 -1 时，您必须关闭 channel 。取消 key 是不够的:您刚刚泄漏了一个套接字。

2. 如果 read() 返回一个正整数，则它是一个读取计数，您忽略了它。您假设您收到了完整的消息。

3. 如果在 Channel 上执行 I/O 时遇到任何 IOException，您必须关闭该 channel 。

可能您不会收到完整的消息，除非您将 sleep 放入。否则，当您将 sleep 放入时，您一次会收到不止一条消息，所以你处理得更快。

添加 sleep 只是修复了你的错误，它本身并没有神奇的特性。