1

大家好！

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

我的期望是这样的——好的，服务器设法读取所有 10000 条消息，即使延迟很短。因此，如果没有这种延迟，服务器可能可以读取更多消息（服务更多客户端）。但是你看，这是错误的。事不宜迟，情况越来越糟。在这里，我描述了我的架构。我当前逻辑中唯一的修改是将接受客户端和读取消息分成 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 条消息。这种奇怪行为的原因是什么？

4

1 回答 1

1

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

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

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

  3. 如果您IOException在 Channel 上进行任何 I/O，则必须关闭该通道。

除非您将睡眠放入其中，否则您可能不会收到整个消息。否则，当您放入睡眠时，您一次会收到多条消息，因此您处理得更快。

添加睡眠只是修复你的错误，它没有自己的神奇属性。

于 2013-05-22T01:43:51.130 回答