转-关于java NIO socket处理粘包 断包问题

TODO / By / 2017-10-25

http://blog.csdn.net/nsdhy/article/details/24249741

 

NIO socket是非阻塞的通讯模式，与IO阻塞式的通讯不同点在于NIO的数据要通过channel放到一个缓存池ByteBuffer中，然后再从这个缓存池中读出数据，而IO的模式是直接从inputstream中read。所以对于NIO，由于存在缓存池的大小限制和网速的不均匀会造成一次读的操作放入缓存池中的数据不完整，便形成了断包问题。同理，如果一次性读入两个及两个以上的数据，则无法分辨两个数据包的界限问题，也就造成了粘包。对于NIO的SocketChannel每次触发OP_READ事件时，发送端不一定仅仅写入了一次，同理，发送端如果一次发送数据包过大，那么发送端的一次写入也可能会被拆分成两次OP_READ事件，所以OP_READ事件和发送端的OP_WRITE事件并不是一一对应的。

第一个问题：对于粘包问题的解决

粘包问题主要是由于数据包界限不清，所以这个问题比较好解决，最好的解决办法就是在发送数据包前事先发送一个int型数据，该数据代表将要发送的数据包的大小，这样接收端可以每次触发OP_READ的时候先接受一个int大小的数据段到缓存池中，然后，紧接着读出后续完整的大小的包，这样就会处理掉粘包问题。因为channel.read()方法不能给读取数据的大小的参数，所以无法手动指定读取数据段的大小。但每次调用channel.read()返回的是他实际读取的大小，这样，思路就有了：首先调整缓存池的大小固定为要读出数据段的大小，这样保证不会过量读出。由于OP_READ和OP_WRITE不是一一对应的，所以一次OP_READ可以While循环调用channel.read()不停读取channel中的数据到缓存池，并捕获其返回值，当返回值累计达到要读取数据段大小时break掉循环，这样保证数据读取充足。所以这样就完美解决粘包问题。

第二个问题：对于断包问题的解决

断包问题主要是由于数据包过量读入时，缓存池结尾处只有半个数据包，channel里还有半个数据包，这样造成了这个包无法处理的问题。这个问题的解决思路是保证每次不过量读入，这样也就不存在断包了。还是因为channel.read()的读取不可控的原因，所以无法从read函数中控制读取大小，还是从缓存池入手。方法是调整缓存池的大小为要读数据的大小，这样就不会断包。

下附某次开发过程的源代码参考：

发送端：

[java] view plain copy

  1. private void sendIntoChannel() {
  3.         Runnable run = new Runnable() {
  4.             @Override
  5.             public void run() {
  7.                 try {
  8.                     ByteArrayOutputStream bOut;
  9.                     ObjectOutputStream out;
  10.                     CBaseDataBean cbdb;
  11.                     ByteBuffer bb = ByteBuffer.allocate(MemCache);
  12.                     while (true) {
  13.                         cbdb = CloudServer.cdsq.read();//Blocking Method
  15.                         //处理自我命令:断开连接 退出线程
  16.                         if (cbdb.getDataType() == CMsgTypeBean.MSG_TYPE_CUTDOWN) {
  17.                             break;
  18.                         }
  20.                         bOut = new ByteArrayOutputStream();
  21.                         out = new ObjectOutputStream(bOut);
  22.                         out.writeObject(cbdb);
  23.                         out.flush();
  25.                         //构造发送数据:整型数据头+有效数据段
  26.                         byte[] arr = bOut.toByteArray();
  27.                         final int ObjLength = arr.length;   //获取有效数据段长度                      
  28.                         bb.clear();
  29.                         bb.limit(IntLength + ObjLength);    //调整缓存池大小
  30.                         bb.putInt(ObjLength);
  31.                         bb.put(arr);
  32.                         bb.position(0);                     //调整重置读写指针
  34.                         SocketChannel channel = (SocketChannel) key.channel();
  35.                         channel.write(bb);
  37.                         out.close();
  38.                         bOut.close();
  39.                     }
  40.                 } catch (IOException ex) {
  41.                 }
  42.             }
  43.         };
  44.         CloudServer.cstp.putNewThread(run);
  45.     }

接收端：

[java] view plain copy

  1. /**
  2.  * 开辟线程分发消息
  3.  */
  4. private void Dispatcher() {
  5.     Runnable run = new Runnable() {
  7.         @Override
  8.         public void run() {
  9.             try {
  10.                 while (true) {
  11.                     selector.selectNow();
  12.                     Thread.sleep(100);
  13.                     Iterator<SelectionKey> itor = selector.selectedKeys().iterator();
  14.                     while (itor.hasNext()) {
  15.                         SelectionKey selKey = itor.next();
  16.                         itor.remove();
  18.                         if (selKey.isValid() && selKey.isAcceptable()) {
  19.                             finshAccept(selKey);
  20.                         }
  22.                         if (selKey.isValid() && selKey.isReadable()) {
  23.                             //消息分发
  24.                             Processer();
  25.                         }
  26.                     }
  27.                 }
  28.             } catch (IOException | InterruptedException ex) {
  29.                 System.out.println(ex.toString());
  30.             }
  31.         }
  32.     };
  33.     CloudServer.cstp.putNewThread(run);
  34. }
  36. /**
  37.  * 消息处理器
  38.  */
  39. private void Processer() {
  40.     ByteBuffer bbInt = ByteBuffer.allocate(IntLength);    //读取INT头信息的缓存池
  41.     ByteBuffer bbObj = ByteBuffer.allocate(MemCache);     //读取OBJ有效数据的缓存池
  42.     SocketChannel channel = (SocketChannel)key.channel();
  43.     ByteArrayInputStream bIn;
  44.     ObjectInputStream in;
  45.     CBaseDataBean cbdb;
  46.     //有效数据长度
  47.     int ObjLength;
  48.     //从NIO信道中读出的数据长度
  49.     int readObj;
  50.     try {
  51.         //读出INT数据头
  52.         while (channel.read(bbInt) == IntLength) {
  53.             //获取INT头中标示的有效数据长度信息并清空INT缓存池
  54.             ObjLength = bbInt.getInt(0);
  55.             bbInt.clear();
  57.             //清空有效数据缓存池设置有效缓存池的大小
  58.             bbObj.clear();
  59.             bbObj.limit(ObjLength);
  61.             //循环读满缓存池以保证数据完整性
  62.             readObj = channel.read(bbObj);
  63.             while (readObj != ObjLength) {
  64.                 readObj += channel.read(bbObj);
  65.             }
  67.             bIn = new ByteArrayInputStream(bbObj.array());
  68.             in = new ObjectInputStream(bIn);
  69.             cbdb = (CBaseDataBean) in.readObject();
  70.             switch (cbdb.getDataType()) {
  71.                 case CMsgTypeBean.MSG_TYPE_COMMAND:
  72.                     rcv_msg_command(cbdb);
  73.                     break;
  74.                 case CMsgTypeBean.MSG_TYPE_CUTDOWN:
  75.                     rcv_msg_cutdown();
  76.                     break;
  77.                 case CMsgTypeBean.MSG_TYPE_VERIFYFILE:
  78.                     rcv_msg_verifyfile(cbdb);
  79.                     break;
  80.                 case CMsgTypeBean.MSG_TYPE_SENDFILE:
  81.                     rcv_msg_sendfile(cbdb);
  82.                     break;
  83.                 case CMsgTypeBean.MSG_TYPE_DISPATCHTASK:
  84.                     rcv_msg_dispchtask(cbdb);
  85.                     break;
  86.             }
  87.             in.close();
  89.         }
  90.     } catch (ClassNotFoundException | IOException ex) {
  91.     }
  92. }