//帮忙看看为什么客户端两百左右! 只返回到一白就越来越变慢
//而且以后就不能回来了! 那里出问题. 不是说完成端口随便写都
//以接受几百到一千吗? 我现在才需要三百个客户端就够了!
//可是为什么不行, 调了几天没有找出原因! 各位高人出手吧!
unit uIOCPComp;interface{$IFDEF VER150}
{$WARN UNSAFE_TYPE OFF}
{$WARN UNSAFE_CODE OFF}
{$WARN UNSAFE_CAST OFF}
{$ENDIF}uses
Windows, Messages, WinSock2, Classes, ScktComp, SysUtils;const
MAX_BUFSIZE = 4096;
WM_CLIENTSOCKET = WM_USER + $2000;type
TCMSocketMessage = packed record
Msg: Cardinal;
Socket: TSocket;
SelectEvent: Word;
SelectError: Word;
Result: Longint;
end; TSocketEvent = (seInitIOPort, seInitSocket, seConnect, seDisconnect,
seListen, seAccept, seWrite, seRead);
TErrorEvent = (eeGeneral, eeSend, eeReceive, eeConnect, eeDisconnect, eeAccept); PPerHandleData = ^TPerHandleData;
TPerHandleData = packed record
Overlapped: OVERLAPPED;
wsaBuffer: WSABUF;
Event: TSocketEvent;
IsUse: Boolean;
Buffer: array [0..MAX_BUFSIZE - 1] of Char;
end; PBlock = ^TBlock;
TBlock = packed record
Data: TPerHandleData;
IsUse: Boolean;
end; EMemoryBuffer = class(Exception);
ESocketError = class(Exception); TCustomSocket = class;
TServerClientSocket = class; TOnDataEvent = function(Socket: TCustomSocket; Data: Pointer; Count: Integer): Integer of object;
TSocketErrorEvent = procedure(Socket: TCustomSocket; ErrorEvent: TErrorEvent; var ErrCode: Integer) of object;
TSocketEventEvent = procedure(Socket: TCustomSocket; SocketEvent: TSocketEvent) of object;
TMemoryBuffer = class
private
FList: TList;
FSocket: TCustomSocket;
function GetCount: Integer;
function GetBlock(const Index: Integer): PBlock;
protected
property Count: Integer read GetCount;
property Blocks[const Index: Integer]: PBlock read GetBlock;
public
constructor Create(ASocket: TCustomSocket); overload;
constructor Create(ASocket: TCustomSocket; BlockCount: Integer); overload;
destructor Destroy; override;
function AllocBlock: PBlock;
procedure RemoveBlock(Block: PBlock);
end; TCustomSocket = class
private
FSocket: TSocket;
FActive: Boolean;
FInitLock: Boolean;
FLock: TRTLCriticalSection;
FOnRead: TOnDataEvent;
FOnErrorEvent: TSocketErrorEvent;
FOnEventEvent: TSocketEventEvent;
function GetRemoteAddress: string;
function GetRemotePort: String;
function GetRemoteHost: string;
procedure DoRead(Data: Pointer; Count: Integer);
protected
procedure SetActive(Value: Boolean); virtual; abstract;
property OnRead: TOnDataEvent read FOnRead write FOnRead;
procedure Event(SocketEvent: TSocketEvent); virtual;
procedure Error(ErrorEvent: TErrorEvent; var ErrCode: Integer); virtual;
property OnErrorEvent: TSocketErrorEvent read FOnErrorEvent write FOnErrorEvent;
property OnEventEvent: TSocketEventEvent read FOnEventEvent write FOnEventEvent;
public
constructor Create(ASocket: TSocket);
destructor Destroy; override;
procedure Close;
procedure Open;
procedure Lock;
procedure UnLock;
function Read(var Buf; Count: Integer): Integer; virtual;
function Write(var Buf; Count: Integer): Integer; virtual;
property SocketHandle: TSocket read FSocket;
property Active: Boolean read FActive write SetActive;
property RemoteHost: string read GetRemoteHost;
property RemotePort: String read GetRemotePort;
property RemoteAddress: string read GetRemoteAddress;
end; TCustomerServerSocket = class(TCustomSocket)
private
FOnClientRead: TOnDataEvent;
FOnClientError: TSocketErrorEvent;
FOnClientEvent: TSocketEventEvent;
protected
function DoClientRead(ASocket: TCustomSocket; AData: Pointer; ACount: Integer): Integer;
procedure ClientSocketError(ASocket: TCustomSocket;
ErrorEvent: TErrorEvent; var ErrCode: Integer);
procedure ClientSocketEvent(ASocket: TCustomSocket; SocketEvent: TSocketEvent);
public
property OnClientRead: TOnDataEvent read FOnClientRead write FOnClientRead;
property OnClientError: TSocketErrorEvent read FOnClientError write FOnClientError;
property OnClientEvent: TSocketEventEvent read FOnClientEvent write FOnClientEvent;
property OnErrorEvent;
property OnEventEvent;
end; TGetSocketEvent = procedure(Socket: TSocket; var ClientSocket: TServerClientSocket) of object;
TServerSocket = class(TCustomerServerSocket)
private
FPort: Integer;
FAddr: TSockAddr;
FAcceptThread: TThread;
FCompletionPort: THandle;
FClients: TList;
FThreads: TList;
FHandle: THandle;
FBuffer: TMemoryBuffer;
FOnGetSocket: TGetSocketEvent;
procedure SetPort(Value: Integer);
function RegisterClient(ASocket: TCustomSocket): Integer;
procedure RemoveClient(ASocket: TCustomSocket);
procedure WMClientClose(var Message: TCMSocketMessage); message WM_CLIENTSOCKET;
procedure WndProc(var Message: TMessage);
function FindClientSocket: TServerClientSocket; overload;
function FindClientSocket(ASocket: TSocket): TCustomSocket; overload;
function GetClientCount: Integer;
function GetClients(const Index: Integer): TServerClientSocket;
protected
procedure InternalOpen;
procedure InternalClose;
procedure SetActive(Value: Boolean); override;
property CompletionPort: THandle read FCompletionPort;
function IsAccept(Socket: TSocket): Boolean; virtual;
public
constructor Create;
destructor Destroy; override;
procedure SetSameClientSocket(SmsCardNo: String);
procedure Accept(ASocket: TSocket; ACompletionPort: THandle);
property Handle: THandle read FHandle;
property Port: Integer read FPort write SetPort;
property ClientCount: Integer read GetClientCount;
property Clients[const Index: Integer]: TServerClientSocket read GetClients;
property OnGetSocket: TGetSocketEvent read FOnGetSocket write FOnGetSocket;
end; TServerClientSocket = class(TCustomSocket)
private
FRegisterIndex: Integer; //接受客户注册索引
FSmsCardNo: String; //车载序列号
FUpdateTime: Cardinal; //接受时间
FBlock: TList;
FBuffer: TMemoryBuffer;
FServerSocket: TServerSocket;
function AllocBlock: PBlock;
function PrepareRecv(Block: PBlock = nil): Boolean;
function WorkBlock(var Block: PBlock; Transfered: DWORD): DWORD;
protected
procedure SetActive(Value: Boolean); override;
public
constructor Create(AServerSocket: TServerSocket; ASocket: TSocket);
destructor Destroy; override;
function Read(var Buf; Count: Integer): Integer; override;
function Write(var Buf; Count: Integer): Integer; override;
property SmsCardNo: String read FSmsCardNo write FSmsCardNo;
property UpdateTime: Cardinal read FUpdateTime write FUpdateTime;
property RegisterIndex: Integer read FRegisterIndex;
end; TSocketThread = class(TThread)
private
FServer: TServerSocket;
public
constructor Create(AServer: TServerSocket);
end; TAcceptThread = class(TSocketThread)
protected
procedure Execute; override;
end; TWorkerThread = class(TSocketThread)
protected
procedure Execute; override;
end;implementationuses RTLConsts;const
SHUTDOWN_FLAG = $FFFFFFFF;
BlockSize: Word = SizeOf(TBlock);var
WSData: TWSAData;{ TMemoryBuffer }
begin
Create(ASocket, 200);
end;constructor TMemoryBuffer.Create(ASocket: TCustomSocket; BlockCount: Integer);
var
I: Integer;
P: PBlock;
begin
inherited Create;
FSocket := ASocket;
FList := TList.Create;
for I := 0 to BlockCount - 1 do
begin
New(P);
FillChar(P^, BlockSize, 0);
FList.Add(P);
end;
end; destructor TMemoryBuffer.Destroy;
var
I: Integer;
begin
for I := FList.Count - 1 downto 0 do
Dispose(PBlock(FList[I]));
FList.Clear;
FList.Free;
inherited Destroy;
end;function TMemoryBuffer.AllocBlock: PBlock;
var
I: Integer;
pbk: PBlock;
begin
FSocket.Lock;
try
Result := nil;
pbk := nil;
for I := 0 to FList.Count - 1 do
begin
pbk := FList[I];
if not pbk.IsUse then
break;
end;
if (pbk = nil) or (pbk.IsUse) then
begin
New(pbk);
FList.Add(pbk);
end;
Result := pbk;
FillChar(Result^.Data, SizeOf(Result^.Data), 0);
Result^.IsUse := True;
finally
FSocket.UnLock;
end;
end;procedure TMemoryBuffer.RemoveBlock(Block: PBlock);
begin
FSocket.Lock;
try
Block.IsUse := False;
finally
FSocket.UnLock;
end;
end;function TMemoryBuffer.GetCount: Integer;
begin
Result := FList.Count;
end;function TMemoryBuffer.GetBlock(const Index: Integer): PBlock;
begin
if (Index >= Count) or (Index <= -1) then
raise EMemoryBuffer.CreateFmt(SListIndexError, [Index])
else
Result := FList[Index];
end;procedure CheckError(ResultCode: Integer; const OP: string);
var
ErrCode: Integer;
begin
if ResultCode <> 0 then
begin
ErrCode := WSAGetLastError;
if (ErrCode <> WSAEWOULDBLOCK) or (ErrCode <> ERROR_IO_PENDING) then
raise ESocketError.CreateFmt(SWindowsSocketError,
[SysErrorMessage(ErrCode), ErrCode, Op]);
end;
end;
begin
inherited Create;
FInitLock := False;
if WSAStartup($0202, WSData) <> 0 then
raise ESocketError.Create(SysErrorMessage(GetLastError));
FSocket := ASocket;
FActive := FSocket <> INVALID_SOCKET;
end;destructor TCustomSocket.Destroy;
begin
//SetActive(False);
WSACleanup;
if FInitLock then
DeleteCriticalSection(FLock);
inherited Destroy;
end;procedure TCustomSocket.Lock;
begin
if not FInitLock then
begin
InitializeCriticalSection(FLock);
FInitLock := True;
end;
EnterCriticalSection(FLock);
end;procedure TCustomSocket.UnLock;
begin
if FInitLock then
LeaveCriticalSection(FLock);
end;procedure TCustomSocket.Close;
begin
SetActive(False);
end;procedure TCustomSocket.Open;
begin
SetActive(True);
end;procedure TCustomSocket.DoRead(Data: Pointer; Count: Integer);
begin
if Assigned(FOnRead) then
FOnRead(Self, Data, Count);
end;procedure TCustomSocket.Error(ErrorEvent: TErrorEvent; var ErrCode: Integer);
begin
if Assigned(FOnErrorEvent) then
FOnErrorEvent(Self, ErrorEvent, ErrCode);
end;procedure TCustomSocket.Event(SocketEvent: TSocketEvent);
begin
if Assigned(FOnEventEvent) then
FOnEventEvent(Self, SocketEvent);
end;function TCustomSocket.GetRemoteAddress: string;
var
SockAddrIn: TSockAddrIn;
Size: Integer;
begin
Result := '';
if not FActive then Exit;
Size := SizeOf(SockAddrIn);
CheckError(getpeername(FSocket, SockAddrIn, Size), 'getpeername');
Result := inet_ntoa(SockAddrIn.sin_addr);
end;function TCustomSocket.GetRemotePort: String;
var
SockAddrIn: TSockAddrIn;
Size: Integer;
begin
Result := '';
if not FActive then Exit;
Size := SizeOf(SockAddrIn);
CheckError(getpeername(FSocket, SockAddrIn, Size), 'getpeername');
Result := IntToStr(SockAddrIn.sin_port);
end;
function TCustomSocket.GetRemoteHost: string;
var
SockAddrIn: TSockAddrIn;
Size: Integer;
HostEnt: PHostEnt;
begin
Result := '';
if not FActive then Exit;
Size := SizeOf(SockAddrIn);
CheckError(getpeername(FSocket, SockAddrIn, Size), 'getpeername');
HostEnt := gethostbyaddr(@SockAddrIn.sin_addr.s_addr, 4, PF_INET);
if HostEnt <> nil then Result := HostEnt.h_name;
end;function TCustomSocket.Read(var Buf; Count: Integer): Integer;
begin
raise ESocketError.Create('Error');
end;function TCustomSocket.Write(var Buf; Count: Integer): Integer;
begin
raise ESocketError.Create('Error');
end;{ TCustomerServerSocket }function TCustomerServerSocket.DoClientRead(ASocket: TCustomSocket;
AData: Pointer; ACount: Integer): Integer;
begin
if not Assigned(FOnClientRead) then
Result := 0 else
Result := FOnClientRead(ASocket, AData, ACount);
end;procedure TCustomerServerSocket.ClientSocketError(ASocket: TCustomSocket;
ErrorEvent: TErrorEvent; var ErrCode: Integer);
begin
if Assigned(FOnClientError) then
FOnClientError(ASocket, ErrorEvent, ErrCode);
end;procedure TCustomerServerSocket.ClientSocketEvent(ASocket: TCustomSocket;
SocketEvent: TSocketEvent);
begin
if Assigned(FOnClientEvent) then
FOnClientEvent(ASocket, SocketEvent);
end;
var
Addr: TSockAddrIn;
AddrLen, Ret, ErrCode: Integer;
ClientWinSocket: TSocket;
ClientSocket: TServerClientSocket;
begin
AddrLen := SizeOf(Addr);
ClientWinSocket := WinSock2.accept(ASocket, Addr, AddrLen);
if ClientWinSocket <> INVALID_SOCKET then
begin
if not Active and not IsAccept(ClientWinSocket) then
begin
closesocket(ClientWinSocket);
Exit;
end;
try
Event(seAccept);
ClientSocket := nil;
ClientSocket := FindClientSocket;
if Assigned(ClientSocket) then
begin
ClientSocket.FSocket := ClientWinSocket;
ClientSocket.FUpdateTime := GetTickCount;
ClientSocket.FActive := True;
ClientSocket.PrepareRecv;
ClientSocket.Event(seConnect);
end;
if not Assigned(ClientSocket) then
ClientSocket := TServerClientSocket.Create(Self, ClientWinSocket);
if Assigned(FOnGetSocket) then
FOnGetSocket(ClientWinSocket, ClientSocket);
except
closesocket(ClientWinSocket);
ErrCode := GetLastError;
Error(eeAccept, ErrCode);
Exit;
end;
Ret := CreateIoCompletionPort(ClientWinSocket, ACompletionPort, DWORD(ClientSocket), 0);
if Ret = 0 then
begin
closesocket(ClientSocket.FSocket);
ClientSocket.FSocket := INVALID_SOCKET;
ClientSocket.FSmsCardNo := '';
ClientSocket.FUpdateTime := 0;
end;
end;
end;constructor TServerSocket.Create;
begin
inherited Create(INVALID_SOCKET);
FBuffer := TMemoryBuffer.Create(Self);
FClients := TList.Create;
FThreads := TList.Create; FPort := 211;
FAcceptThread := nil;
FCompletionPort := 0;
IsMultiThread := True;
FHandle := Classes.AllocateHWnd(WndProc);
end;destructor TServerSocket.Destroy;
begin
//SetActive(False);
FThreads.Free;
FClients.Free;
Classes.DeallocateHWnd(FHandle);
FBuffer.Free;
inherited Destroy;
end;function TServerSocket.FindClientSocket(ASocket: TSocket): TCustomSocket;
var
I: Integer;
begin
Lock;
try
Result := nil;
for I := FClients.Count - 1 downto 0 do
begin
Result := FClients[I];
if ASocket = Result.SocketHandle then Break;
end;
finally
UnLock;
end;
end;function TServerSocket.FindClientSocket: TServerClientSocket;
var
I: Integer;
ClientSocket: TServerClientSocket;
begin
Lock;
try
Result := nil;
for I := FClients.Count - 1 downto 0 do
begin
ClientSocket := TServerClientSocket(FClients[I]);
with ClientSocket do
if FSocket = INVALID_SOCKET then
begin
Result := ClientSocket;
Break;
end;
end;
finally
UnLock;
end;
end;
function TServerSocket.GetClientCount: Integer;
begin
Result := FClients.Count;
end;function TServerSocket.GetClients(const Index: Integer): TServerClientSocket;
begin
Result := FClients[Index];
end;procedure TServerSocket.InternalClose;
procedure CloseObject(var Handle: THandle);
begin
if Handle <> 0 then
begin
CloseHandle(Handle);
Handle := 0;
end;
end;var
I: Integer;
Thread: TThread;
begin
Lock;
try
for I := FThreads.Count - 1 downto 0 do
begin
Thread := FThreads[I];
Thread.Terminate;
PostQueuedCompletionStatus(FCompletionPort, 0, 0, Pointer(SHUTDOWN_FLAG));
end;
FThreads.Clear;
if FAcceptThread <> nil then FAcceptThread.Terminate;
while FClients.Count > 0 do
TObject(FClients.Last).Free;
FClients.Clear;
if FSocket <> INVALID_SOCKET then
begin
Event(seDisconnect);
closesocket(FSocket);
FSocket := INVALID_SOCKET;
end;
if FCompletionPort <> 0 then CloseObject(FCompletionPort);
finally
UnLock;
end;
end;procedure TServerSocket.InternalOpen;
var
I: Integer;
Thread: TThread;
SystemInfo: TSystemInfo;
begin
Lock;
try
try
FCompletionPort := CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0, 2);
if FCompletionPort = 0 then
raise ESocketError.Create(SysErrorMessage(GetLastError)); Event(seInitIOPort);
GetSystemInfo(SystemInfo);
for I := 0 to SystemInfo.dwNumberOfProcessors - 1 do
begin
Thread := TWorkerThread.Create(Self);
FThreads.Add(Thread);
end;
FSocket := WSASocket(PF_INET, SOCK_STREAM, 0, nil, 0, WSA_FLAG_OVERLAPPED);
if FSocket = INVALID_SOCKET then
raise ESocketError.Create(SysErrorMessage(GetLastError));
Event(seInitSocket); FillChar(FAddr, SizeOf(FAddr), 0);
FAddr.sin_family := AF_INET;
FAddr.sin_port := htons(FPort);
FAddr.sin_addr.S_addr := INADDR_ANY;
CheckError(bind(FSocket, @FAddr, SizeOf(FAddr)), 'bind'); Event(seListen);
CheckError(listen(FSocket, 50), 'listen');
FAcceptThread := TAcceptThread.Create(Self);
except
InternalClose;
raise;
end;
finally
UnLock;
end;
end;function TServerSocket.IsAccept(Socket: TSocket): Boolean;
begin
Result := True;
end;function TServerSocket.RegisterClient(ASocket: TCustomSocket): Integer;
begin
Lock;
try
Result := -1;
if FClients.IndexOf(ASocket) = -1 then
begin
Result := FClients.Add(ASocket);
WSAAsyncSelect(ASocket.SocketHandle, FHandle, WM_CLIENTSOCKET, FD_CLOSE);
end;
finally
UnLock;
end;
end;procedure TServerSocket.RemoveClient(ASocket: TCustomSocket);
var
Index: Integer;
begin
Lock;
try
Index := FClients.IndexOf(ASOcket);
if Index <> -1 then
FClients.Delete(Index);
finally
UnLock;
end;
end;procedure TServerSocket.SetActive(Value: Boolean);
begin
if FActive = Value then Exit;
if Value then InternalOpen else InternalClose;
FActive := Value;
end;procedure TServerSocket.SetPort(Value: Integer);
begin
if Active then
raise ESocketError.Create('Cann''t change port');
FPort := Value;
end;procedure TServerSocket.SetSameClientSocket(SmsCardNo: String);
var
I: Integer;
ClientSocket: TServerClientSocket;
begin
Lock;
try
for I := FClients.Count - 1 downto 0 do
begin
ClientSocket := TServerClientSocket(FClients[I]);
if Trim(ClientSocket.FSmsCardNo) = Trim(SmsCardNo) then
begin
ClientSocket.FSocket := INVALID_SOCKET;
ClientSocket.FActive := False;
Break;
end;
end;
finally
UnLock;
end;
end;procedure TServerSocket.WMClientClose(var Message: TCMSocketMessage);
var
ClientSocket: TCustomSocket;
begin
ClientSocket := FindClientSocket(Message.Socket);
if Assigned(ClientSocket) then
begin
closesocket(ClientSocket.FSocket);
ClientSocket.FSocket := INVALID_SOCKET;
end;
end;procedure TServerSocket.WndProc(var Message: TMessage);
begin
try
Dispatch(Message);
except
if Assigned(ApplicationHandleException) then
ApplicationHandleException(Self);
end;
end;
2. 为什么继承自TServerSocket并且用了非阻塞消息模式? 既然已经用了非阻塞消息模式再用完成端口是不是有脱裤子放屁的嫌疑?3. 完成端口用于socket编程的一般流程是:
1. 初始化创建一个完成端口和一批工作线程
2. 将Accept或connect的已连通的socket绑定到这个完成端口
3. 通过这个绑定的socket投递一个WSARecv（接收）或WSASend（发送）请求
4. 在工作线程中处理这个WSARecv(已收到全部或部分数据)或WSASend(已发送成功全部或部分数据)完成后的工作我在你的代码中并没有看到这些操作.
而ServerSocket则是通过socket 1.1的Recv和Send来读写的，所以你的代码根本不是完成端口应用。只是创建了一批浪费资源用的线程和完成端口操作而已
//现在贴上了
{ TServerClientSocket }constructor TServerClientSocket.Create(AServerSocket: TServerSocket;
ASocket: TSocket);
begin
inherited Create(ASocket);
FUpdateTime := GetTickCount;
FServerSocket := AServerSocket;
FBuffer := FServerSocket.FBuffer;
FBlock := TList.Create;
FRegisterIndex := FServerSocket.RegisterClient(Self);
FOnRead := FServerSocket.OnClientRead;
OnErrorEvent := FServerSocket.ClientSocketError;
OnEventEvent := FServerSocket.ClientSocketEvent;
PrepareRecv;
Event(seConnect);
end;destructor TServerClientSocket.Destroy;
var
I: Integer;
begin
FServerSocket.RemoveClient(Self);
closesocket(FSocket);
for I := FBlock.Count - 1 downto 0 do
FBuffer.RemoveBlock(FBlock[I]);
FBlock.Clear;
FBlock.Free;
inherited Destroy;
end;procedure TServerClientSocket.SetActive(Value: Boolean);
var
Linger: TLinger;
begin
if FActive = Value then Exit;
if not Value then
begin
if FSocket <> INVALID_SOCKET then
begin
Event(seDisconnect);
FillChar(Linger, SizeOf(Linger), 0);
setsockopt(FSocket, SOL_SOCKET, SO_LINGER, @Linger, Sizeof(Linger));
closesocket(FSocket);
FSocket := INVALID_SOCKET;
end;
end else
raise ESocketError.Create('当前socket不支持连接操作');
FActive := Value;
end; function TServerClientSocket.AllocBlock: PBlock;
var
I: Integer;
begin
for I := 0 to FBlock.Count - 1 do
begin
Result := FBlock[I];
if not Result.Data.IsUse then
begin
Result.Data.IsUse := True;
Exit;
end;
end;
Result := FBuffer.AllocBlock;
FBlock.Add(Result);
Result.Data.IsUse := True;
end;function TServerClientSocket.Read(var Buf; Count: Integer): Integer;
begin
{ 读操作由DoReceive触发OnRead进行读 }
raise ESocketError.Create('读操作错误');
end;function TServerClientSocket.Write(var Buf; Count: Integer): Integer;
var
Block: PBlock;
ErrCode: Integer;
Flags, BytesSend: Cardinal;
begin
Result := Count;
if Result = 0 then Exit;
Block := AllocBlock;
with Block^.Data do
begin
Flags := 0;
Event := seWrite;
wsaBuffer.buf := @Buf;
wsaBuffer.len := Result;
if SOCKET_ERROR = WSASend(FSocket, @wsaBuffer, 1, BytesSend, Flags, @Overlapped, nil) then
begin
ErrCode := WSAGetLastError;
if ErrCode <> ERROR_IO_PENDING then
begin
Result := SOCKET_ERROR;
Error(eeSend, ErrCode);
end;
end;
end;
end;function TServerClientSocket.PrepareRecv(Block: PBlock = nil): Boolean;
var
ErrCode: Integer;
Flags, Transfer: Cardinal;
begin
Result := False;
if not Assigned(Block) then
Block := AllocBlock;
with Block^.Data do
begin
Flags := 0;
Transfer := 0;
Event := seRead;
FillChar(Buffer, SizeOf(Buffer), 0);
FillChar(Overlapped, SizeOf(Overlapped), 0);
wsaBuffer.buf := Buffer;
wsaBuffer.len := MAX_BUFSIZE;
Result := SOCKET_ERROR <> WSARecv(FSocket, @wsaBuffer, 1, Transfer, Flags, @Overlapped, nil);
if not Result then
begin
ErrCode := WSAGetLastError;
Result := ErrCode = ERROR_IO_PENDING;
if not Result then
begin
Block.Data.IsUse := False;
Error(eeReceive, ErrCode);
end;
end;
end;
end;const
RESPONSE_UNKNOWN = $0001;
RESPONSE_SUCCESS = $0002;
RESPONSE_FAIL = $FFFF;
function TServerClientSocket.WorkBlock(var Block: PBlock; Transfered: DWORD): DWORD;
var
ErrCode: Integer;
Flag, BytesSend: Cardinal;
begin
Result := RESPONSE_SUCCESS;
UpdateTime := GetTickCount;
with Block^.Data do
try
case Block^.Data.Event of
seRead:
begin
Self.Event(seRead);
DoRead(@Buffer, Transfered);
if not Self.PrepareRecv(Block) then
Result := RESPONSE_FAIL;
end;
seWrite:
begin
Self.Event(seWrite);
Dec(wsaBuffer.len, Transfered);
if wsaBuffer.len <= 0 then
begin
{ 发送完成，将Block置空，返回到FBlock的可使用的缓区中 }
Block.Data.IsUse := False;
Block := nil;
end else
begin
{ 数据还没发送完成，继续发送 }
Flag := 0;
Inc(wsaBuffer.buf, Transfered);
FillChar(Overlapped, SizeOf(Overlapped), 0);
if SOCKET_ERROR = WSASend(FSocket, @wsaBuffer, 1, BytesSend,
Flag, @Overlapped, nil) then
begin
ErrCode := WSAGetLastError;
if ErrCode <> ERROR_IO_PENDING then
Error(eeSend, ErrCode);
end;
end;
end;
end;
except
Result := RESPONSE_FAIL;
end;
end;{ TSocketThread }constructor TSocketThread.Create(AServer: TServerSocket);
begin
FServer := AServer;
inherited Create(False);
FreeOnTerminate := True;
end;{ TAcceptThread }procedure TAcceptThread.Execute;
begin
with FServer do
while not Terminated and Active do
Accept(SocketHandle, CompletionPort);
end;{ TWorkerThread }procedure TWorkerThread.Execute;
var
Block: PBlock;
Transfered: DWORD;
ClientSocket: TServerClientSocket;
begin
while not Terminated do
begin
Block := nil;
Transfered := 0;
ClientSocket := nil;
if not GetQueuedCompletionStatus(FServer.CompletionPort, Transfered,
DWORD(ClientSocket), POverlapped(Block), INFINITE) then
begin
if Assigned(ClientSocket) then FreeAndNil(ClientSocket);
Continue;
end;
{ 通知结束 }
if Cardinal(Block) = SHUTDOWN_FLAG then break;
if not FServer.Active then break;
{ 客户可能超时?? 或是断开连接，I/O失败 }
if Transfered = 0 then
begin
{if Assigned(ClientSocket) then
FreeAndNil(ClientSocket);}
Continue;
end;
case ClientSocket.WorkBlock(Block, Transfered) of
RESPONSE_UNKNOWN:
{ 操作未知的话，应该返回给客户端：...不应该Close....保留 }
if Assigned(ClientSocket) then FreeAndNil(ClientSocket);
RESPONSE_FAIL:
if Assigned(ClientSocket) then FreeAndNil(ClientSocket);
end;
end;
end;
end.
多来这么几次那么线程池就空了. 此时你就再也收不到任何消息了.