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

Posted 2020-09-13 学无止境

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.     }