DELPHI高性能大容量SOCKET并发（四）：粘包、分包、解包

DELPHI高性能大容量SOCKET并发（四）：粘包、分包、解包

粘包


使用TCP长连接就会引入粘包的问题，粘包是指发送方发送的若干包数据到接收方接收时粘成一包，从接收缓冲区看，后一包数据的头紧接着前一包数据的尾。粘包可能由发送方造成，也可能由接收方造成。TCP为提高传输效率，发送方往往要收集到足够多的数据后才发送一包数据，造成多个数据包的粘连。如果接收进程不及时接收数据，已收到的数据就放在系统接收缓冲区，用户进程读取数据时就可能同时读到多个数据包。

粘包一般的解决办法是制定通讯协议，由协议来规定如何分包解包。


分包


在IOCPDemo例子程序中，我们分包的逻辑是先发一个长度，然后紧接着是数据包内容，这样就可以把每个包分开。

应用层数据包格式如下：



应用层数据包格式   
数据包长度Len：Cardinal（4字节无符号整数） 数据包内容，长度为Len 
IOCPSocket分包处理主要代码，我们收到的数据都是在TSocketHandle.ProcessIOComplete方法中处理：




[delphi] view plaincopy在CODE上查看代码片派生到我的代码片
01.procedure TSocketHandle.ProcessIOComplete(AIocpRecord: PIocpRecord;  
02.  const ACount: Cardinal);  
03.begin  
04.  case AIocpRecord.IocpOperate of  
05.    ioNone: Exit;  
06.    ioRead: //收到数据  
07.    begin  
08.      FActiveTime := Now;  
09.      ReceiveData(AIocpRecord.WsaBuf.buf, ACount);  
10.      if FConnected then  
11.        PreRecv(AIocpRecord); //投递请求  
12.    end;  
13.    ioWrite: //发送数据完成，需要释放AIocpRecord的指针  
14.    begin  
15.      FActiveTime := Now;  
16.      FSendOverlapped.Release(AIocpRecord);  
17.    end;  
18.    ioStream:  
19.    begin  
20.      FActiveTime := Now;  
21.      FSendOverlapped.Release(AIocpRecord);  
22.      WriteStream; //继续发送流  
23.    end;  
24.  end;  
25.end;  

如果是收到数据，则调用ReceiveData函数，ReceiveData主要功能是把数据的写入流中，然后调用Process分包。FInputBuf是一个内存流（FInputBuf: TMemoryStream），内存流的每次写入会造成一次内存分配，如果要获得更高的效率，可以替换为内存池等更好的内存管理方式。还有一种更好的解决方案是规定每次发包的大小，如每个包最大不超过64K，哪么缓冲区的最大大小可以设置为128K（缓存两个数据包），这样就可以每次创建对象时一次分配好，减少内存分配次数，提高效率。（内存的分配和释放比内存的读写效率要低） 




[delphi] view plaincopy在CODE上查看代码片派生到我的代码片
01.procedure TSocketHandle.ReceiveData(AData: PAnsiChar; const ALen: Cardinal);  
02.begin  
03.  FInputBuf.Write(AData^, ALen);  
04.  Process;  
05.end;  

Process则根据收到的数据进行分包逻辑，如果不够一个包，则继续等待接收数据，如果够一个或多个包，则循环调用Execute函数进行处理，代码如下： 




[delphi] view plaincopy在CODE上查看代码片派生到我的代码片
01.procedure TSocketHandle.Process;  
02.var  
03.  AData, ALast, NewBuf: PByte;  
04.  iLenOffset, iOffset, iReserveLen: Integer;  
05.  
06.  function ReadLen: Integer;  
07.  var  
08.    wLen: Word;  
09.    cLen: Cardinal;  
10.  begin  
11.    FInputBuf.Position := iOffset;  
12.    if FLenType = ltWord then  
13.    begin  
14.      FInputBuf.Read(wLen, SizeOf(wLen));  
15.      //wLen := ntohs(wLen);  
16.      Result := wLen;  
17.    end  
18.    else  
19.    begin  
20.      FInputBuf.Read(cLen, SizeOf(cLen));  
21.      //cLen := ntohl(cLen);  
22.      Result := cLen;  
23.    end;  
24.  end;  
25.begin  
26.  case FLenType of  
27.    ltWord, ltCardinal:  
28.    begin  
29.      if FLenType = ltWord then  
30.        iLenOffset := 2  
31.      else  
32.        iLenOffset := 4;  
33.      iReserveLen := 0;  
34.      FPacketLen := 0;  
35.      iOffset := 0;  
36.      if FPacketLen <= 0 then  
37.      begin  
38.        if FInputBuf.Size < iLenOffset then Exit;  
39.        FInputBuf.Position := 0; //移动到最前面  
40.        FPacketLen := ReadLen;  
41.        iOffset := iLenOffset;  
42.        iReserveLen := FInputBuf.Size - iOffset;  
43.        if FPacketLen > iReserveLen then //不够一个包的长度  
44.        begin  
45.          FInputBuf.Position := FInputBuf.Size; //移动到最后，以便接收后续数据  
46.          FPacketLen := 0;  
47.          Exit;  
48.        end;  
49.      end;  
50.      while (FPacketLen > 0) and (iReserveLen >= FPacketLen) do //如果数据够长，则处理  
51.      begin //多个包循环处理  
52.        AData := Pointer(Longint(FInputBuf.Memory) + iOffset); //取得当前的指针  
53.        Execute(AData, FPacketLen);  
54.        iOffset := iOffset + FPacketLen; //移到下一个点  
55.        FPacketLen := 0;  
56.        iReserveLen := FInputBuf.Size - iOffset;  
57.        if iReserveLen > iLenOffset then //剩下的数据  
58.        begin  
59.          FPacketLen := ReadLen;  
60.          iOffset := iOffset + iLenOffset;  
61.          iReserveLen := FInputBuf.Size - iOffset;  
62.          if FPacketLen > iReserveLen then //不够一个包的长度，需要把长度回退  
63.          begin  
64.            iOffset := iOffset - iLenOffset;  
65.            iReserveLen := FInputBuf.Size - iOffset;  
66.            FPacketLen := 0;  
67.          end;  
68.        end  
69.        else //不够长度字节数  
70.          FPacketLen := 0;  
71.      end;  
72.      if iReserveLen > 0 then //把剩下的自己缓存起来  
73.      begin  
74.        ALast := Pointer(Longint(FInputBuf.Memory) + iOffset);  
75.        GetMem(NewBuf, iReserveLen);  
76.        try  
77.          CopyMemory(NewBuf, ALast, iReserveLen);  
78.          FInputBuf.Clear;  
79.          FInputBuf.Write(NewBuf^, iReserveLen);  
80.        finally  
81.          FreeMemory(NewBuf);  
82.        end;  
83.      end  
84.      else  
85.      begin  
86.        FInputBuf.Clear;  
87.      end;  
88.    end;  
89.  else  
90.    begin  
91.      FInputBuf.Position := 0;  
92.      AData := Pointer(Longint(FInputBuf.Memory)); //取得当前的指针  
93.      Execute(AData, FInputBuf.Size);  
94.      FInputBuf.Clear;  
95.    end;  
96.  end;  
97.end;  



解包


由于我们应用层数据包既可以传命令也可以传数据，因而针对每个包我们进行解包，分出命令和数据分别处理，因而每个Socket服务对象都需要解包，我们解包的逻辑是放在TBaseSocket.DecodePacket中，命令和数据的包格式为：



命令长度Len：Cardinal（4字节无符号整数） 命令 数据 
这里和第一版公布的代码不同，这版的代码对命令进行了编码，采用UTF-8编码，代码如下：




[delphi] view plaincopy在CODE上查看代码片派生到我的代码片
01.function TBaseSocket.DecodePacket(APacketData: PByte;  
02.  const ALen: Integer): Boolean;  
03.var  
04.  CommandLen: Integer;  
05.  UTF8Command: UTF8String;  
06.begin  
07.  if ALen > 4 then //命令长度为4字节，因而长度必须大于4  
08.  begin  
09.    CopyMemory(@CommandLen, APacketData, SizeOf(Cardinal)); //获取命令长度  
10.    Inc(APacketData, SizeOf(Cardinal));  
11.    SetLength(UTF8Command, CommandLen);  
12.    CopyMemory(PUTF8String(UTF8Command), APacketData, CommandLen); //读取命令  
13.    Inc(APacketData, CommandLen);  
14.    FRequestData := APacketData; //数据  
15.    FRequestDataLen := ALen - SizeOf(Cardinal) - CommandLen; //数据长度  
16.    FRequest.Text := Utf8ToAnsi(UTF8Command); //把UTF8转为Ansi  
17.    Result := True;  
18.  end  
19.  else  
20.    Result := False;   
21.end;  

具体每个协议可以集成Execute方法，调用DecodePacket进行解包，然后根据命令进行协议逻辑处理，例如TSQLSocket主要代码如下： 




[delphi] view plaincopy在CODE上查看代码片派生到我的代码片
01.{* SQL查询SOCKET基类 *}  
02.TSQLSocket = class(TBaseSocket)  
03.private  
04.  {* 开始事务创建TADOConnection，关闭事务时释放 *}  
05.  FBeginTrans: Boolean;  
06.  FADOConn: TADOConnection;  
07.protected  
08.  {* 处理数据接口 *}  
09.  procedure Execute(AData: PByte; const ALen: Cardinal); override;  
10.  {* 返回SQL语句执行结果 *}  
11.  procedure DoCmdSQLOpen;  
12.  {* 执行SQL语句 *}  
13.  procedure DoCmdSQLExec;  
14.  {* 开始事务 *}  
15.  procedure DoCmdBeginTrans;  
16.  {* 提交事务 *}  
17.  procedure DoCmdCommitTrans;  
18.  {* 回滚事务 *}  
19.  procedure DoCmdRollbackTrans;  
20.public  
21.  procedure DoCreate; override;  
22.  destructor Destroy; override;  
23.  {* 获取SQL语句 *}  
24.  function GetSQL: string;  
25.  property BeginTrans: Boolean read FBeginTrans;  
26.end;  

Exceute是调用DecodePacket进行解包，然后获取命令分别调用不同的命令处理逻辑，代码如下： 




[delphi] view plaincopy在CODE上查看代码片派生到我的代码片
01.procedure TSQLSocket.Execute(AData: PByte; const ALen: Cardinal);  
02.var  
03.  sErr: string;  
04.begin  
05.  inherited;  
06.  FRequest.Clear;  
07.  FResponse.Clear;  
08.  try  
09.    AddResponseHeader;  
10.    if ALen = 0 then  
11.    begin  
12.      DoFailure(CIPackLenError);  
13.      DoSendResult;  
14.      Exit;  
15.    end;  
16.    if DecodePacket(AData, ALen) then  
17.    begin  
18.      FResponse.Clear;  
19.      AddResponseHeader;  
20.      case StrToSQLCommand(Command) of  
21.        scLogin:  
22.        begin  
23.          DoCmdLogin;  
24.          DoSendResult;  
25.        end;  
26.        scActive:  
27.        begin  
28.          DoSuccess;  
29.          DoSendResult;  
30.        end;  
31.        scSQLOpen:  
32.        begin  
33.          DoCmdSQLOpen;  
34.        end;  
35.        scSQLExec:  
36.        begin  
37.          DoCmdSQLExec;  
38.          DoSendResult;  
39.        end;  
40.        scBeginTrans:  
41.        begin  
42.          DoCmdBeginTrans;  
43.          DoSendResult;  
44.        end;  
45.        scCommitTrans:  
46.        begin  
47.          DoCmdCommitTrans;  
48.          DoSendResult;  
49.        end;  
50.        scRollbackTrans:  
51.        begin  
52.          DoCmdRollbackTrans;  
53.          DoSendResult;  
54.        end;  
55.      else  
56.        DoFailure(CINoExistCommand, 'Unknow Command');  
57.        DoSendResult;  
58.      end;  
59.    end  
60.    else  
61.    begin  
62.      DoFailure(CIPackFormatError, 'Packet Must Include 

');  
63.      DoSendResult;  
64.    end;  
65.  except  
66.    on E: Exception do //发生未知错误，断开连接  
67.    begin  
68.      sErr := RemoteAddress + ':' + IntToStr(RemotePort) + CSComma + 'Unknow Error: ' + E.Message;  
69.      WriteLogMsg(ltError, sErr);  
70.      Disconnect;  
71.    end;  
72.  end;  
73.end;  



更详细代码见示例代码的IOCPSocket单元。


V1版下载地址：http://download.csdn.net/detail/sqldebug_fan/4510076，需要资源10分，有稳定性问题，可以作为研究稳定性用；
V2版下载地址：http://download.csdn.net/detail/sqldebug_fan/5560185，不需要资源分，解决了稳定性问题和提高性能；免责声明：此代码只是为了演示IOCP编程，仅用于学习和研究，切勿用于商业用途。水平有限，错误在所难免，欢迎指正和指导。邮箱地址：fansheng_hx@163.com。