首页 > 代码库 > NIO框架之MINA源码解析（四）：粘包与断包处理及编码与解码

NIO框架之MINA源码解析（四）：粘包与断包处理及编码与解码

2024-09-16 21:17:25   218人阅读

1、粘包与段包

 

粘包：指TCP协议中，发送方发送的若干包数据到接收方接收时粘成一包，从接收缓冲区看，后一包数据的头紧接着前一包数据的尾。
造成的可能原因：

    发送端需要等缓冲区满才发送出去，造成粘包

    接收方不及时接收缓冲区的包，造成多个包接收

 

断包：也就是数据不全，比如包太大，就把包分解成多个小包，多次发送，导致每次接收数据都不全。

 

2、消息传输的格式

 

消息长度+消息头+消息体  即前N个字节用于存储消息的长度，用于判断当前消息什么时候结束。

消息头+消息体    即固定长度的消息，前几个字节为消息头，后面的是消息头。

在MINA中用的是

消息长度+消息体 即前4个字节用于存储消息的长度，用于判断当前消息什么时候结束。

 

3、编码与解码

 

   在网络中，信息的传输都是通过字节的形式传输的，而我们在编写自己的代码时，则都是具体的对象，那么要想我们的对象能够在网络中传输，就需要编码与解码。

 

   编码：即把我们的消息编码成二进制形式，能以字节的形式在网络中传输。

   解码：即把我们收到的字节解码成我们代码中的对象。

   在MINA中对象的编码与解码用的都是JDK提供的ObjectOutputStream来实现的。

 

4、MINA中消息的处理实现

 

消息的接受处理，我们常用的是TCP协议，而TCP协议会分片的，在下面的代码中，具体功能就是循环从通道里面读取数据，直到没有数据可读，或者buffer满了，然后就把接受到的数据发给解码工厂进行处理。

 

4.1、消息的接收

 

 

[java] view plain copy
 
 print?
  1. //class AbstractPollingIoProcessor  
  2. private void read(S session) {  
  3.         IoSessionConfig config = session.getConfig();  
  4.         int bufferSize = config.getReadBufferSize();  
  5.         IoBuffer buf = IoBuffer.allocate(bufferSize);  
  6.   
  7.         final boolean hasFragmentation = session.getTransportMetadata().hasFragmentation();  
  8.   
  9.         try {  
  10.             int readBytes = 0;  
  11.             int ret;  
  12.   
  13.             try {  
  14.                 //是否有分片 tcp传输会有分片，即把大消息分片成多个小消息再传输  
  15.                 if (hasFragmentation) {  
  16.             //read方法非阻塞，没有读到数据的时候返回0  
  17.                     while ((ret = read(session, buf)) > 0) {  
  18.                         readBytes += ret;  
  19.                         //buffer 满了  
  20.                         if (!buf.hasRemaining()) {  
  21.                             break;  
  22.                         }  
  23.                     }  
  24.                 } else {  
  25.                     ret = read(session, buf);  
  26.   
  27.                     if (ret > 0) {  
  28.                         readBytes = ret;  
  29.                     }  
  30.                 }  
  31.             } finally {  
  32.                 buf.flip();  
  33.             }  
  34.   
  35.             if (readBytes > 0) {  
  36.                 IoFilterChain filterChain = session.getFilterChain();  
  37.         //处理消息  
  38.                 filterChain.fireMessageReceived(buf);  
  39.                 buf = null;  
  40.   
  41.                 if (hasFragmentation) {  
  42.                     if (readBytes << 1 < config.getReadBufferSize()) {  
  43.                         session.decreaseReadBufferSize();  
  44.                     } else if (readBytes == config.getReadBufferSize()) {  
  45.                         session.increaseReadBufferSize();  
  46.                     }  
  47.                 }  
  48.             }  
  49.   
  50.             if (ret < 0) {  
  51.                 scheduleRemove(session);  
  52.             }  
  53.         } catch (Throwable e) {  
  54.             if (e instanceof IOException) {  
  55.                 if (!(e instanceof PortUnreachableException)  
  56.                         || !AbstractDatagramSessionConfig.class.isAssignableFrom(config.getClass())  
  57.                         || ((AbstractDatagramSessionConfig) config).isCloseOnPortUnreachable()) {  
  58.                     scheduleRemove(session);  
  59.                 }  
  60.             }  
  61.   
  62.             IoFilterChain filterChain = session.getFilterChain();  
  63.             filterChain.fireExceptionCaught(e);  
  64.         }  
  65.     }  



 

 

 

 

 

 

4.2、解码与编码

 

 

[java] view plain copy
 
 print?
  1. //class AbstractIoBuffer  
  2.  public Object getObject(final ClassLoader classLoader) throws ClassNotFoundException {  
  3.     //首先判断当前buffer中消息长度是否完整，不完整的话直接返回  
  4.         if (!prefixedDataAvailable(4)) {  
  5.             throw new BufferUnderflowException();  
  6.         }  
  7.   
  8.     //消息长度  
  9.         int length = getInt();  
  10.         if (length <= 4) {  
  11.             throw new BufferDataException("Object length should be greater than 4: " + length);  
  12.         }  
  13.   
  14.         int oldLimit = limit();  
  15.     //limit到消息结尾处  
  16.         limit(position() + length);  
  17.         try {  
  18.             ObjectInputStream in = new ObjectInputStream(asInputStream()) {  
  19.                 @Override  
  20.                 protected ObjectStreamClass readClassDescriptor() throws IOException, ClassNotFoundException {  
  21.                     int type = read();  
  22.                     if (type < 0) {  
  23.                         throw new EOFException();  
  24.                     }  
  25.                     switch (type) {  
  26.                     case 0: // NON-Serializable class or Primitive types  
  27.                         return super.readClassDescriptor();  
  28.                     case 1: // Serializable class  
  29.                         String className = readUTF();  
  30.                         Class<?> clazz = Class.forName(className, true, classLoader);  
  31.                         return ObjectStreamClass.lookup(clazz);  
  32.                     default:  
  33.                         throw new StreamCorruptedException("Unexpected class descriptor type: " + type);  
  34.                     }  
  35.                 }  
  36.   
  37.                 @Override  
  38.                 protected Class<?> resolveClass(ObjectStreamClass desc) throws IOException, ClassNotFoundException {  
  39.                     String name = desc.getName();  
  40.                     try {  
  41.                         return Class.forName(name, false, classLoader);  
  42.                     } catch (ClassNotFoundException ex) {  
  43.                         return super.resolveClass(desc);  
  44.                     }  
  45.                 }  
  46.             };  
  47.             return in.readObject();  
  48.         } catch (IOException e) {  
  49.             throw new BufferDataException(e);  
  50.         } finally {  
  51.             limit(oldLimit);  
  52.         }  
  53.     }  
  54.   
  55. //判断当前消息是否完整   
  56. public boolean prefixedDataAvailable(int prefixLength, int maxDataLength) {  
  57.         if (remaining() < prefixLength) {  
  58.             return false;  
  59.         }  
  60.   
  61.         int dataLength;  
  62.         switch (prefixLength) {  
  63.         case 1:  
  64.             dataLength = getUnsigned(position());  
  65.             break;  
  66.         case 2:  
  67.             dataLength = getUnsignedShort(position());  
  68.             break;  
  69.         case 4:  
  70.             dataLength = getInt(position());  
  71.             break;  
  72.         default:  
  73.             throw new IllegalArgumentException("prefixLength: " + prefixLength);  
  74.         }  
  75.   
  76.         if (dataLength < 0 || dataLength > maxDataLength) {  
  77.             throw new BufferDataException("dataLength: " + dataLength);  
  78.         }  
  79.     //判断当前消息是否完整   
  80.         return remaining() - prefixLength >= dataLength;  
  81.     }  
  82.   
  83. //编码  
  84.  public IoBuffer putObject(Object o) {  
  85.         int oldPos = position();  
  86.         skip(4); // Make a room for the length field.预留4个字节用于存储消息长度  
  87.         try {  
  88.             ObjectOutputStream out = new ObjectOutputStream(asOutputStream()) {  
  89.                 @Override  
  90.                 protected void writeClassDescriptor(ObjectStreamClass desc) throws IOException {  
  91.                     try {  
  92.                         Class<?> clz = Class.forName(desc.getName());  
  93.                         if (!Serializable.class.isAssignableFrom(clz)) { // NON-Serializable class  
  94.                             write(0);  
  95.                             super.writeClassDescriptor(desc);  
  96.                         } else { // Serializable class  
  97.                             write(1);  
  98.                             writeUTF(desc.getName());  
  99.                         }  
  100.                     } catch (ClassNotFoundException ex) { // Primitive types  
  101.                         write(0);  
  102.                         super.writeClassDescriptor(desc);  
  103.                     }  
  104.                 }  
  105.             };  
  106.             out.writeObject(o);  
  107.             out.flush();  
  108.         } catch (IOException e) {  
  109.             throw new BufferDataException(e);  
  110.         }  
  111.   
  112.         // Fill the length field  
  113.         int newPos = position();  
  114.         position(oldPos);  
  115.     //存储消息长度  
  116.         putInt(newPos - oldPos - 4);  
  117.         position(newPos);  
  118.         return this;  
  119.     }  



 

 

 

 

 

 

4.3、断包与粘包处理

 

[java] view plain copy
 
 print?
  1. // class CumulativeProtocolDecoder  
  2.  public void decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception {  
  3.     //是否有分片，tcp 有分片  
  4.         if (!session.getTransportMetadata().hasFragmentation()) {  
  5.             while (in.hasRemaining()) {  
  6.                 if (!doDecode(session, in, out)) {  
  7.                     break;  
  8.                 }  
  9.             }  
  10.   
  11.             return;  
  12.         }  
  13.   
  14.     // 1、断包处理  
  15.     // 2、处理粘包  
  16.         boolean usingSessionBuffer = true;  
  17.     //session中是否有断包情况（上次处理后），断包保存在session中  
  18.         IoBuffer buf = (IoBuffer) session.getAttribute(BUFFER);  
  19.         // If we have a session buffer, append data to that; otherwise  
  20.         // use the buffer read from the network directly.  
  21.         if (buf != null) {//有断包，则把当前包拼接到断包里面  
  22.             boolean appended = false;  
  23.             // Make sure that the buffer is auto-expanded.  
  24.             if (buf.isAutoExpand()) {  
  25.                 try {  
  26.                     buf.put(in);  
  27.                     appended = true;  
  28.                 } catch (IllegalStateException e) {  
  29.                     // A user called derivation method (e.g. slice()),  
  30.                     // which disables auto-expansion of the parent buffer.  
  31.                 } catch (IndexOutOfBoundsException e) {  
  32.                     // A user disabled auto-expansion.  
  33.                 }  
  34.             }  
  35.   
  36.             if (appended) {  
  37.                 buf.flip();  
  38.             } else {  
  39.                 // Reallocate the buffer if append operation failed due to  
  40.                 // derivation or disabled auto-expansion.  
  41.                 buf.flip();  
  42.                 IoBuffer newBuf = IoBuffer.allocate(buf.remaining() + in.remaining()).setAutoExpand(true);  
  43.                 newBuf.order(buf.order());  
  44.                 newBuf.put(buf);  
  45.                 newBuf.put(in);  
  46.                 newBuf.flip();  
  47.                 buf = newBuf;  
  48.   
  49.                 // Update the session attribute.  
  50.                 session.setAttribute(BUFFER, buf);  
  51.             }  
  52.         } else {  
  53.             buf = in;  
  54.             usingSessionBuffer = false;  
  55.         }  
  56.   
  57.     //2 粘包处理，可能buffer中有多个消息，需要多次处理（解码）每个消息，直到消息处理完，或者剩下的消息不是一个完整的消息或者buffer没有数据了  
  58.   
  59.         for (;;) {  
  60.             int oldPos = buf.position();  
  61.             boolean decoded = doDecode(session, buf, out);  
  62.             if (decoded) {//解码 成功  
  63.                 if (buf.position() == oldPos) {  
  64.                     throw new IllegalStateException("doDecode() can‘t return true when buffer is not consumed.");  
  65.                 }  
  66.         //buffer空了  
  67.                 if (!buf.hasRemaining()) {//buffer没有数据了  
  68.                     break;  
  69.                 }  
  70.             } else {//剩下的消息不是一个完整的消息，断包出现了  
  71.                 break;  
  72.             }  
  73.         }  
  74.   
  75.         // if there is any data left that cannot be decoded, we store  
  76.         // it in a buffer in the session and next time this decoder is  
  77.         // invoked the session buffer gets appended to  
  78.         if (buf.hasRemaining()) {//剩下的消息不是一个完整的消息，断包出现了  
  79.         //如果断包已经保存在session中，则更新buffer，没有的话，就把剩下的断包保存在session中  
  80.             if (usingSessionBuffer && buf.isAutoExpand()) {  
  81.                 buf.compact();  
  82.             } else {  
  83.                 storeRemainingInSession(buf, session);  
  84.             }  
  85.         } else {  
  86.             if (usingSessionBuffer) {  
  87.                 removeSessionBuffer(session);  
  88.             }  
  89.         }  
  90.     }  



 

 

 

 

 

 

 

[java] view plain copy
 
 print?
    1. //class  ObjectSerializationDecoder  
    2.  protected boolean doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception {  
    3.     //首先判断当前buffer中消息长度是否完整，不完整的话直接返回  
    4.         if (!in.prefixedDataAvailable(4, maxObjectSize)) {  
    5.             return false;  
    6.         }  
    7.   
    8.         out.write(in.getObject(classLoader));  
    9.         return true;  
    10.     }  

NIO框架之MINA源码解析（四）：粘包与断包处理及编码与解码


联系
我们