DELPHI高性能大容量SOCKET并发

1、DELPHI高性能大容量 SOCKET并发（一）: IOCP完成端口例子介绍SQL查询、上传、下载等协议实例子主要包括iocp控件封装、服务端实现、传输协议和日志、控制、 现，并包括一些初步的性能测试结果。服务端：界面截图如下：应TO提供服务和桌面方式运行，桌面方式可直接打开程序，方便日常调试，可以使用命令行注册或卸载服务, 在 CMD 中输入 D:DEMOIOCPDemoBinIOCPDemoSvr.exe -install来注册服务，在 CMD 输入D:DEMOIOCPDemoBi nIOCPDemoSvr.exe -u nin stall来卸载服务。客户端：界面截图如下：13主要实现了服

2、务端日志查看，服务端协议类表查看，SQL语句执行协议，上传、下载协议实现，其中对上传、下载实现了一个多线程同时传，用于测试服务器并发性能。性能：支持超过2000个链接及以上同时上传文件，不过每个连接上传速度只有1到2K。支持超过2W个连接同时在线传输命令。单实例上传下载测试结果：*HcDvF1rwiMot口卜，:KB 1:?rq.o |* “KmT-i? IftS J屮仍 ijcXP ( KB岳Wtt (KSJ5 .云主砥Sft1535.07661.6454 9155JJil.sai719.&11041.917B7J397S2.6a42244J77a.1429.4-1込瞞i料经垃naui2H7

3、J27.jU16S125.a1253.625J.ZS327M1.4139WWI5& B277D537W.J7inII&4175M 加帅i削n2644孜關MIR.弼B9T8 理1钏飾7&*1 314U.RT1375412923. 316757 02B5593W.6751230e6EL22059.421134KlS2427.MLd241706.711604714107.07血Ml射:，.：，卜 SLHttML：U1吨“上Gff两应丈小: KB：|12Mb 11B i KB?$)t v FJJ1 l?(KfV5 1S网(B/S池(KB/5云1635.-I75SL5 7418152C28*5.15BS

4、9 471MJ5xl4UJflJg1S3.J622aJH2.Hg1皿就2316瞬侶125937.256J63-Uia.aa649ieO.B61M心424H.4S111404巧1 H756翦 TKLZS33421J481.415123D332.&7Z&4M 旳432. MWM2M40.1119549.N从测试结果可以看出随着发送包增大，速度变快。这里存在一个风险，就是SOCKET传输失败的次数也会增加。(二) ：IOCP完成端口控件封装IOCP完成端口介绍：完成端口模型是 Windows平台下SOCKET端口模型最为复杂的一种 I/O模型。如果一个应用程序需要同 时管理为数众多的套接字，而且希望

5、随着系统内安装的CPU数量的增多，应用程序的性能也可以线性提升，采用完成端口模型，往往可以达到最佳的系统性能。完成端口可以管理成千上万的连接，长连接传文件可以支持 5000个以上，长连接命令交互可以支持20000个以上。这么大并发的连接，更需要考虑的是应用场景，按照100M的网卡传输速度12.5MB/S，如果是5000个传文件连接，则每个连接能分到的速度2.56KB/S ;如果是20000个命令交互连接，则每个连接分到的吞吐量是655B/S，这种速度的吞吐量对很多应用是不满足，这时就要考虑加大网卡的传输速度或实现水平扩展， 这个我们后续会介绍。完成端口是由系统内核管理多个线程之间的切换，比外部

6、实现线程池性能要高，CPU利用率上内核和用户态可以达到1:1，很多应用线程池是无法达到的。因此同等连接数的情况下，完成端口要比INDY的TCPServer传输速度要快，吞吐量更高。要使用完成端口，主要是以下三个函数的使用：CreateloCompletionPort、GetQueuedCompletionStatus、PostQueuedCompletio nStatus。CreateIoCompletio nPort的功能是：1、创建一个完成端口对象；2、将一个句柄和完成端口关联在一起；GetQueuedCompleti on Status是获取完成端口状态，是阻塞式调用，在指定时间内如果没

7、有事件通知，会一直等待；PostQueuedCompleti on Status用于向完成端口投递一个完成事件通知。function CreateIoCompletio nPort(FileHa ndle, Existi ngCompletio nPort: THan dle; Completi on Key, NumberOfCo ncurren tThreads: DWORD): THa ndle; stdcall;NumberOfC on curre ntThreads参数定义了在一个完成端口上，同时允许执行的线程数量。将NumberOfConcurrentThreads设为0表示每个处

8、理器各自负责一个线程的运行，为完成端口提供服务，避免过于频繁的线程场景切换。因此可以使用下列语句来创建一个完成 端口 FlocpHandle := CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0, 0);执行线程个数创建完成端口后，就可以将套接字句柄与对象关联在一起，这时就需要创建工作者线程，以便在完成端口 收到数据后，为完成端口提供处理数据线程。到底创建多少个线程为完成端口服务，这个是完成端口最为复杂 的一方面，创建多了线程会造成频繁的线程场景切换；创建少了线程如果某一个处理非常耗时，如连接数据库、读写文件，又会造成完成端口拥塞，因此这个参

9、数需要提供设置，并根据最终的应用场景反复测试得出一个结 果。一般的经验值是设置为CPU的个数*2+4 ；IOCP完成端口一般使用步骤1、创建一个完成端口；2、判断系统内安装了多少个处理器；3、创建工作者线程；4、创建一个SOCKET套接字开始监听；5、使用Accept接收连接；6、 调用CreateIoCompletionPort将连接和完成端口绑定在一起；7、投递接收数据请求8、 工作者线程调用 GetQueuedCompletio nStatus获取事件通知，处理数据；IOCP控件核心代码第1步到第4步实现代码：delphi view pla in copy1. procedure TIo

10、cpServer . Open;2. var3. WsaData: TWsaData;4. iNumberOfProcessors, i, iWorkThreadCount: Integer;5. WorkThread: TWorkThread;6. Addr: TSockAddr;7. begin8. if WSAStartup( $0202 , WsaData) 0 then / 初始化 SOCKET9. raise ESocketError . Create(GetLastWsaErrorStr);10. FIocpHandle := CreateloCompletionPort(INV

11、ALID_HANDLE_VALUE,0, 0, 0);/ 创建一个完成端口11. if FIocpHandle =0 then12. raise ESocketError . Create(GetLastErrorStr);13. FSocket := WSASocket(PF_INET, SOCK_STREAM,0, nil , 0, WSA_FLAG_OVERLAPPED); / 仓U建一个 SOCKET句柄14.15.if FSocket = INVALID SOCKET thenraise ESocketError . Create(GetLastWsaErrorStr);16.Fil

12、lChar(Addr, SizeOf(Addr),0);17.Addr.sin_family := AF_INET;18.Addr.sin_port := htons(FPort);19.Addr.sin_addr . S_addr := htonl(INADDR_ANY);以指定只监听某一个IP地址/在任何地址上监听，如果有多块网卡，会每块都监听，也可20.ifbind(FSocket, Addr, SizeOf(Addr) then /把SOCKET句柄绑定端口21.22.ifraise ESocketError . Create(GetLastWsaErrorStr); listen(F

13、Socket, MaxInt) 0 then23.24.iNumberOfProcessors := GetCPUCount;/获取CPU个数25.iWorkThreadCount := iNumberOfProcessors *2 + 4; /由于服务器处理可能比较费时间，因此线程设为raise ESocketError . Create(GetLastWsaErrorStr);CPU*2+426.if iWorkThreadCount FMaxWorkThrCounttheniWorkThreadCount := FMinWorkThrCount;29.30.for i :=0 to iW

14、orkThreadCount -do/创建工作者线程31.begin32.WorkThread := TWorkThread.Create(Self, True);iWorkThreadCount := FMaxWorkThrCount;33.34.WorkThread . Resume;35.end;36.FAcceptThreadPool . Active := True;/启动监听线程池37.FAcceptThread := TAcceptThread.Create(Self, True);/启动监听线程38.FAcceptThread . Resume;39.end;FWorkThre

15、ads . Add(WorkThread);第5步和第6步实现代码:delphi view pla in cedure TIocpServer . AcceptClient;2.var3. ClientSocket: TSocket;4. begin5. ClientSocket := WSAAccept(FSocket,nil , nil , nil , 0);/ 接收连接6. if ClientSocket INVALID_SOCKETthen7. begin8.if not FActive then9. begin10. closesocket(ClientSocket

16、);11. Exit;12. end;13. FAcceptThreadPool . PostSocket(ClientSocket);/这里使用线程池主要作用是为了判断发送的第一个字节身份标识，用来是判断协议类型14. end;15. end;FAcceptThreadPool 是一个用完成端口实现的线程池TIocpThreadPool ，主要是有TCheckThread 来判断完成端口的返回，并检测是否6S内有发送标志位上来，主要实现过程是响应 On Co nnect事件，并在On Co nnect 事件中判断是否允许连接。delphi view plai ncopy1. procedur

17、e TIocpServer . CheckClient( const ASocket: TSocket);2. var3. SocketHandle: TSocketHandle;4. iIndex: Integer;5. ClientSocket: PClientSocket;6. begin7. SocketHandle := nil ;8. if not DoConnect(ASocket, SocketHandle)then / 如果不允许连接，则退出9. begin10. closesocket(ASocket);11. Exit;12. end;13. FSocketHandles

18、 . Lock; / 加到列表中14. try15. iIndex := FSocketHandles. Add(SocketHandle);16. ClientSocket := FSocketHandles. ItemsiIndex;17. finally18. FSocketHandles . UnLock;19. end;20. if CreateIoCompletionPort(ASocket, FIOCPHandle, DWORD(ClientSocket),0) =0 then / 将连接和完成端口绑定在一起21.begin22.DoError(CreateIoCompletio

19、nPort,GetLastWsaErrorStr)23.FSocketHandles.Lock; II如果投递到列表中失败，则删除24.try25.FSocketHandles.Delete(iIndex);26.finally27.FSocketHandles.UnLock;28.end;29.end30.else31.begin32.SocketHandle.PreRecv( nil ); II投递接收请求33.end;34. end;delphiview pla in cedure TDMDispatchCenter . lcpSvrConnect( constASoc

20、ket: Cardinal;2.var AAllowConnect: Boolean;var SocketHandle: TSocketHandle);3.var4.BaseSocket: TBaseSocket;5.cchFlag: Char;6.7.function GetSocket( const AEnable: Boolean; BaseSocketClass: TBaseSocketClass): TBaseSocket;8.begin9.if AEnable then10.Result := BaseSocketClass. Create(lcpSvr, ASocket)11.e

21、lse12.Result :=nil ;13.end;14.begin15.if (GIniOptions. MaxSocketCount 0) and (IcpSvr.SocketHandles . Count = GIniOptions.MaxSocketCount)then16.begin17.AAllowConnect := False;18.Exit;19.end;20.if IcpSvr . ReadChar(ASocket, chFlag,6* 1000) then / 必须在 6S 内收到标志21.begin22.case TSocketFlag(Byte(chFlag)of2

22、3.sfSQL: BaseSocket := GetSocket(GlniOptions.SQLProtocol, TSQLSocket);24.sfUpload: BaseSocket := GetSocket(GIniOptions.UploadProtocol, TUploadSocket);25.sfDownload: BaseSocket := GetSocket(GIniOptions.DownloadProtocol, TDownloadSocket);26.sfControl: BaseSocket := GetSocket(GIniOptions.ControlProtoco

23、l, TControlSocket);27.sfLog: BaseSocket := GetSocket(GIniOptions.LogProtocol, TLogSocket);28.else29.BaseSocket :=nil ;30.end;31.if BaseSocket nil then32.begin33.SocketHandle := BaseSocket;34.dWriteLogMsg(ltDebug, Format(Client Connect, Local Address: %s:%d; Remote Address: %s:%35.SocketHandle. Local

24、Address, SocketHandle.LocalPort, SocketHandle. RemoteAddress, SocketHandleRemotePort);36.WriteLogMsg(ltDebug, Format(Client Count: %d,IcpSvr . SocketHandles . Count +1);37.AAllowConnect := True;38.end39.else40.AAllowConnect := False;41.end42.else43.AAllowConnect := False;44.end;其中ReadChar函数是用于判断指定时间

25、是否有数据上来，函数实现过程用Select函数检测：delphiview pla in copy1.function TIocpServer . ReadChar( const ASocket: TSocket;var AChar: Char;const ATimeOutMS: Integer): Boolean;2.var3.iRead: Integer;4.begin5.Result := CheckTimeOut(ASocket, ATimeOutMS);6.if Result then7.begin8.iRead := recv(ASocket, AChar,1, 0);9.Resul

26、t := iRead =1;10.end;11.end;12.13.function TIocpServer . CheckTimeOut( const ASocket: TSocket;14.const ATimeOutMS: Integer): Boolean;15.var16.tmTo: TTimeVal;17.FDRead: TFDSet;18.begin19.FillChar(FDRead, SizeOf(FDRead),0);20.FDRead . fd_count :=1;21.FDRead . fd_array 0 := ASocket;22.tmTo . tv_sec :=

27、ATimeOutMSdiv 1000 ;23.tmTo . tv_usec := (ATimeOutMSmod 1000) *1000 ;24.Result := Select(0, FDRead, nil , nil , tmTO) =1;25.end;第7步实现代码接收连接之后要投递接收数据请求，实现代码：delphiview pla in copy|1.procedure TSocketHandle . PreRecv(AlocpRecord: PlocpRecord);2.var3.iFlags, iTransfer: Cardinal;4.iErrCode: Integer;5.be

28、gin6.if not Assigned(AIocpRecord)then7.begin8.New(AIocpRecord);9.AIocpRecord. WsaBuf. buf := FlocpRecvBuf;10.AIocpRecord. WsaBuf. len := MAX_IOCPBUFSIZE;11.FIocpRecv := AIocpRecord;12.end;13.AIocpRecord . Overlapped . Internal :=0;14.AlocpRecord . Overlapped . InternalHigh :=0;15.AlocpRecord . Overl

29、apped . Offset :=0;16.AIocpRecord . Overlapped . OffsetHigh :=0;17.AIocpRecord . Overlapped . hEvent :=0;18./AlocpRecord.WsaBuf.buf := FlocpRecvBuf;19./AlocpRecord.WsaBuf.len := MAX_IOCPBUFSIZE;20.AIocpRecord . locpOperate := ioRead;21.iFlags :=0;22.if WSARecv(FSocket, AIocpRecord. WsaBuf, 1, iTrans

30、fer, iFlags, AIocpRecord.Overlapped,23.nil ) = SOCKET_ERROR then24.begin25.iErrCode := WSAGetLastError;26.if iErrCode = WSAECONNRESET then / 客户端被关闭27.FConnected := False;28.if iErrCode ERROR_IO_PENDINGthen / 不抛出异常，触发异常事件29.begin30.FlocpServer. DoError( WSARecv , GetLastWsaErrorStr);31.ProcessNetErro

31、r(iErrCode);32.end;33.end;34.end;第8步实现代码：delphiview pla in copy1.function TlocpServer . WorkClient: Boolean;2.var3.ClientSocket: PClientSocket;4.locpRecord: PlocpRecord;5.iWorkCount: Cardinal;6.begin7.locpRecord :=nil ;8.iWorkCount :=0;9.ClientSocket :=nil ;10.Result := False;11.if not GetQueuedComp

32、letionStatus(FIocpHandle, iWorkCount, DWORD(ClientSocket),12.POverlapped(IocpRecord), INFINITE)then / 此处有可能多个线程处理同一个SocketHandle 对象，因此需要加锁13.begin /客户端异常断开14.if Assigned(ClientSocket)and Assigned(ClientSocket.SocketHandle)then15.begin16.ClientSocket. SocketHandle . FConnected := False;17.Exit;18.end

33、;19.end;20.if Cardinal(IocpRecord) = SHUTDOWN_FLAGthen21.Exit;22.if not FActive then23.Exit;9. )then4.65.66.ConnectedResult := True;if Assigned(ClientSocket)and Assigned(ClientSocke

34、t . SocketHandle) thenbeginif ClientSocket . SocketHandle . Connected thenbeginif iWorkCount 0 thenbegintryClientSocket. Lock . Enter;tryif Assigned(ClientSocket . SocketHandle) thenClientSocket. SocketHandle . ProcesslOComplete(locpRecord, iWorkCount);finallyClientSocket. Lock . Leave;end;if Assign

35、ed(ClientSocket . SocketHandle) and ( not ClientSocket . SocketHandlebeginClientSocket. Lock . Enter;tryif Assigned(ClientSocket . SocketHandle) thenFreeSocketHandle(ClientSocket. SocketHandle);finallyClientSocket. Lock . Leave;end ;end;excepton E: Exception doDoError(ProcessIOComplete , E . Message

36、);end;endelse /如果完成个数为0,且状态为接收，则释放连接beginif IocpRecord . IocpOperate = ioRead thenbeginClientSocket. Lock . Enter;tryif Assigned(ClientSocket . SocketHandle) thenFreeSocketHandle(ClientSocket. SocketHandle);finallyClientSocket. Lock . Leave;end;endelse67. DoError(+ GetLastErrorStr);68. end;WorkClien

37、t , WorkCount = 0, Code:+ IntToStr(GetLastError) +,Message69. end70. else/断开连接71. begin72. Clientsocket.Lock. Enter;73. try74. if Assigned(ClientSocket . SocketHandle) then75. FreeSocketHandle(ClientSocket. SocketHandle);76. finally77. ClientSocket. Lock. Leave;78. end;79. end;80. end81. else /WorkC

38、ount 为0表示发生了异常，记录日志82. DoError( GetQueuedCompletionStatus , Return SocketHandle nil);83. end;第8步主要是使用 GetQueuedCompletionStatus函数来获取完成端口事件通知，如果有事件完成，则可以通过lpCompletionKey来知道是哪个句柄有数据收到。这里有个技巧是我们在绑定的时候使用如下语句：CreateloCompleti on Port(ASocket, FIOCPHa ndle,DWORD(ClientSocket), 0)，用lpCompletionKey 来传递了一个指

39、针，这其中就包含了一个锁和服务对象，定 义结构如下：delphi view plaincopy1. *客户端对象和锁*2. TClientSocket = record3. Lock: TCriticalSection;4. SocketHandle: TSocketHandle;5. end;6. PClientSocket = TClientSocket;因而我们在收到事件通知后，就可以直接调用TSocketHa ndle对象来完成分包解包以及后续的逻辑处理。 这样一个完成端口的整体骨架就有了，后续还有收发数据、分包解包和业务逻辑处理，以及对象和锁分离等细 节处理。更详细代码见示例代码的l

40、OCPSocket单元。(三) ：接收、发送、缓存接收完成端口是结合重叠端口一起使用的，在接收数据之前，我们需要投递一个接收请求，使用fun ctionWSARecv(s: TSocket; IpBuffers: LPWSABUF; dwBufferCou nt: DWORD;var IpNumberOfBytesRecvd,IpFlags: DWORD; lpOverlapped: LPWSAOVERLAPPED;lpCompletio nRouti ne:LPWSAOVERLAPPED_COMPLETION_ROUTINE): In teger函数，接收连接后，就需要投递一个请求，代码如下

41、：delphiview pla in cedure TSocketHandle . PreRecv(AlocpRecord: PIocpRecord);2.var3.iFlags, iTransfer: Cardinal;4.iErrCode: Integer;5.begin6.if not Assigned(AIocpRecord)then7.begin8.New(AIocpRecord);9.AIocpRecord. WsaBuf. buf := FlocpRecvBuf;10.AIocpRecord. WsaBuf. len := MAX_IOCPBUFSIZE;11.

42、FIocpRecv := AIocpRecord;12.end;13.AIocpRecord . Overlapped . Internal :=0;14.AIocpRecord . Overlapped . InternalHigh :=0;15.AIocpRecord . Overlapped . Offset :=0;16.AIocpRecord . Overlapped . OffsetHigh :=0;17.AIocpRecord . Overlapped . hEvent :=0;18.AIocpRecord . IocpOperate := ioRead;19.iFlags :=

43、0;20.if WSARecv(FSocket, AIocpRecord. WsaBuf, 1, iTransfer, iFlags, AIocpRecord.Overlapped,21.nil ) = SOCKET_ERROR then22.begin23.iErrCode := WSAGetLastError;24.if iErrCode = WSAECONNRESET then / 客户端被关闭25.FConnected := False;26.if iErrCode ERROR_IO_PENDINGthen / 不抛出异常，触发异常事件27.begin28.FIocpServer. D

44、oError( WSARecv , GetLastWsaErrorStr);29.ProcessNetError(iErrCode);30.end;31.end;32.end;WSARecv会返回错误，可以帮助我们定位网络问题，如连接断了，具体函数代码如下：delphiview plai ncopy1. procedure TSocketHandle . ProcessNetError( const AErrorCode: Integer);2. begin3. if AErrorCode = WSAECONNRESET then / 客户端断开连接4.FConnected := False5

45、.else if AErrorCode = WSAEDISCON then / 客户端断开连接6.FConnected := False7.else if AErrorCode = WSAENETDOWNthen/网络异常8.FConnected := False9.else if AErrorCode = WSAENETRESETthen/心跳包拦截到的异常10.FConnected := False11.else if AErrorCode = WSAESHUTDOWNthen/连接被关闭12.FConnected := False13.else if AErrorCode = WSAET

46、IMEDOUTthen/连接断掉或者网络异常14.FConnected := False;15.end;WSARecv函数还有另外一个作用，如果客户端断开连接了，WSARecv会收到一个0字节的返回，我们可以根据这个释放连接对象。如果客户端直接拔网线，WSARecv是检测不到，这是我们要根据超时来检测。接收连接之后我们投递一次接收数据请求，然后收到数据处理后，再投递下一次请求，因而我们相当于每 次只有一个接收请求，我们只需要定义一个接收结构体和一个固定大小的缓存就可以了，类似：delphiview plai ncopy f1. *保存投递请求的地址信息*2. FlocpRecv: PlocpR

47、ecord;3. FIocpRecvBuf:array 0. MAX_IOCPBUFSIZE-1 of Char;接收的数据我们一起放在一个内存流中，主要代码如下：delphi view pla in copy1. procedure TSocketHandle. ProcesslOComplete(AlocpRecord: PIocpRecord;2. const ACount: Cardinal);3. begin4. case AIocpRecord . IocpOperateof5. ioNone: Exit;6. ioRead:/收到数据7. begin8. FActiveTime

48、:= Now;9. ReceiveData(AIocpRecord. WsaBuf. buf, ACount);10. if FConnected then11. PreRecv(AIocpRecord);/投递请求12. end;13. ioWrite:/发送数据完成，需要释放AIocpRecord 的指针14. begin15.FActiveTime := Now;16.FSendOverlapped.Release(AIocpRecord);17.end;18.ioStream:19.begin20.FActiveTime := Now;21.FSendOverlapped.Releas

49、e(AIocpRecord);22.WriteStream;/继续发送流23.end;25. end;26.26. procedure TSocketHandle . ReceiveData(AData: PAnsiChar;const ALen: Cardinal);27. begin28. FlnputBuf . Write(ADataA, ALen);29. Process;30. end;有很多完成端口为了提高效率，会对接收队列使用内存池，以减少内存的分配和释放次数，使用内存流会有频繁的申请和释放操作，也容易造成碎片，我们这里没有使用内存池，而是使用FastMM来加快速度。FastMM在多线程和内存碎片方面做的比DELPHI自带的内存管理器要强，一般能提高2到3倍的性能。发送发送和接收是类似的，使用fun ctio n WSASe nd(s: TSocket; IpBuffers: LPWSABUF; dwBufferCou nt:DWORD; var IpNumberOfBytesSe nt: DWORD; dwFIags: DWORD; IpOverlappe