iocpcomponent.pas

一个用delphi封装的IOCP（完成端口）控件

unit IOCPComponent;
{
   根据网络上的例子修改封装的
   里面有详细的注解.
   QQ:46494153
   2009-03-17
}
interface

uses
  Windows,Messages, SysUtils, Classes,WinSock2,Math,Forms;

const WM_IOCP = WM_USER + 300;//消息 目前未使用
    DATA_BUFSIZE = 1024 * 1024; //数据缓冲区大小
    IPLen = 15;  //IP地址长度 V4 = 255.255.255.255
type
  TIOCPException = Exception;
   //完成端口数据结构
  PLinkInfo = ^TLinkInfo;
  TLinkInfo = record //连接信息
    Skt: TSocket;
    IP: array [0..IPLen -1] of Char;
    Port: integer;
  end;

  LPVOID = Pointer;
  LPPER_IO_OPERATION_DATA = ^ PER_IO_OPERATION_DATA ;
  PER_IO_OPERATION_DATA = packed record
    Overlapped: OVERLAPPED;
    DataBuf: TWSABUF;
    Buffer: array [0..DATA_BUFSIZE -1] of CHAR; //接收缓存区
    SendPointer: PChar;     //发送的指针头
    SendPointerSize: DWORD; //发送指针大小   用于发送完毕释放申请的内存
    BytesSEND: DWORD;
    BytesRECV: DWORD;
  end;

  LPPER_HANDLE_DATA = ^ PER_HANDLE_DATA;//单句柄数据结构 
  PER_HANDLE_DATA = packed record
    Socket: TSocket;
  end;   
  //事件声明
  TIOCPReadEvent = procedure(ASkt: TSocket;AData: PChar;ADataLen: Integer) of object; //读取数据
  TIOCPConnectEvent = procedure(ASkt: TSocket;ALinkInfo: PLinkInfo;var Accept: Boolean) of object;//连接
  TIOCPDisConnectEvent = procedure(ASkt: TSocket) of object;//断开连接

  TIOCPCustomComponent = class(TComponent) //公共类
  private
    FHandle: THandle;
    FOnRead: TIOCPReadEvent;
    FOnDisConnect: TIOCPDisConnectEvent;
    FOnConnect: TIOCPConnectEvent;
    FCompletionPort: THandle;
    function GetHandle: THandle;
    procedure SetOnRead(const Value: TIOCPReadEvent);
    procedure SetOnDisConnect(const Value: TIOCPDisConnectEvent);
    procedure SetOnConnect(const Value: TIOCPConnectEvent);
    function GetCompletionPort: THandle;
  protected
    Flags: integer; 
    PerHandleData: LPPER_HANDLE_DATA;
    PerIoData: LPPER_IO_OPERATION_DATA;
    procedure WMProc(var Msg: TMessage);Message WM_IOCP;
    procedure DoRead(ASkt: TSocket;AData: PChar;ADataLen: Integer);
    procedure DoConnect(ASkt: TSocket;ALinkInfo: PLinkInfo;var Accept: Boolean); virtual;
    procedure DoDisConnect(ASkt: TSocket); virtual;
    procedure WriteLog(ALogText: string);
  public
    property Handle: THandle read GetHandle;
    property CompletionPort: THandle read GetCompletionPort;
    constructor Create(AOwner: TComponent);override;
    destructor Destroy; override;
    function SendData(ASkt: TSocket; AData: PChar; ADataLen: Int64): Boolean;
  published
    property OnRead: TIOCPReadEvent read FOnRead write SetOnRead;
    property OnConnect: TIOCPConnectEvent read FOnConnect write SetOnConnect;
    property OnDisConnect: TIOCPDisConnectEvent read FOnDisConnect write SetOnDisConnect;
  end;
  
  TIOCPServer = class;  //对象相互引用声明

  TThreadServer = class(TThread) //服务监听线程
  private
    FIOCPServer: TIOCPServer;
  protected
    procedure Execute; override;
  public
    constructor Create(Owner:TIOCPServer);
    destructor Destroy; override;
  end;

  TServerWorkerThread = class(TThread) //服务工作者线程
  private
    FCompletionPort: THandle;
    FOwner: TIOCPCustomComponent;
  protected
    procedure Execute; override;
  public
    constructor Create(ACompletionPort: THandle;AOwner:TIOCPCustomComponent);
    destructor Destroy; override;
  end;

  TIOCPServer = class(TIOCPCustomComponent) //服务端
  private
    FThreadServer: TThreadServer;
    FLinkList: TStringList;
    FActive: Boolean;
    FPort: Integer;
    function GetLinkCount: integer;
    procedure SetActive(const Value: Boolean);
    procedure SetPort(const Value: Integer);
    property OnConnect;
  protected
    procedure DoConnect(ASkt: TSocket;ALinkInfo: PLinkInfo;var Accept: Boolean); override;
    procedure DoDisConnect(ASkt: TSocket); override;
  public
    constructor Create(AOwner: TComponent);override;
    destructor Destroy; override;
    Property LinkCount: integer read GetLinkCount; //当前连接个数
    function BroadCast(AData: PChar;ADataLen: Int64): Integer;//广播
  published
    property Active: Boolean read FActive write SetActive default False; //激活
    property Port: Integer read FPort write SetPort default 6666;
  end;

  TIOCPClient = class(TIOCPCustomComponent) //客户端
  private
    FSkt: TSocket;
    FPort: Integer;
    FConnected: Boolean;
    FHost: string;
    procedure SetPort(const Value: Integer);
    procedure SetConnected(const Value: Boolean);
    procedure SetHost(const Value: string);
    property OnConnect;
    //function SendData(ASkt: TSocket; AData: PChar; ADataLen: Int64): Boolean;
  protected
    procedure DoDisConnect(ASkt: TSocket); override; //断开连接
  public
    constructor Create(AOwner: TComponent);override;
    destructor Destroy; override;
    procedure Connect;
    function SendData(AData: PChar; ADataLen: Int64): Boolean;
    property Connected: Boolean read FConnected write SetConnected;
  published
    property Host: string read FHost write SetHost;
    property Port: Integer read FPort write SetPort default 6666;
  end;

procedure Register;

implementation
{$R *.dcr}
procedure Register;
begin
  RegisterComponents('Standard', [TIOCPServer,TIOCPClient]);
end;

{ TIOCPCustomComponent }

constructor TIOCPCustomComponent.Create(AOwner: TComponent);
var
  wsData:TWsaData;
begin
  inherited;
  FHandle := Handle;

  if WSAStartUp($202, wsData) <> 0 then
  begin
   WSACleanup();
   WriteLog('加载WinSock2失败.');
   exit;
  end;
  WriteLog('加载WinSock2成功,版本:' + Inttostr(wsData.wVersion) + '  ' + wsData.szDescription);
end;

destructor TIOCPCustomComponent.Destroy;
begin
  WSACleanup();
  inherited;
end;

procedure TIOCPCustomComponent.DoConnect(ASkt: TSocket;ALinkInfo: PLinkInfo;var Accept: Boolean);
begin
  if Assigned(FOnConnect) then
  FOnConnect(ASkt,ALinkInfo,Accept);
end;

procedure TIOCPCustomComponent.DoDisConnect(ASkt: TSocket);
begin
  if Assigned(FOnDisConnect) then
  FOnDisConnect(ASkt);
end;

procedure TIOCPCustomComponent.DoRead(ASkt: TSocket; AData: PChar;
  ADataLen: Integer);
var
  DataBuf: array of Char;
begin
  SetLength(DataBuf,ADataLen+1);
  Move(AData^,DataBuf[0],ADataLen);
  if Assigned(FOnRead) then
  FOnRead(ASkt,PChar(@(DataBuf[0])),ADataLen);
  //FreeAndNil(DataBuf);
end;

function TIOCPCustomComponent.GetCompletionPort: THandle;
var
  LocalSI: TSystemInfo;
  I: integer;
  ServerWorkerThread: TServerWorkerThread;
begin
  if FCompletionPort=0 then
  begin
    FCompletionPort:=CreateIOCompletionPort(INVALID_HANDLE_VALUE,0,0,0);
    WriteLog('创建一个完成端口。');
    GetSystemInfo(LocalSI);
    //根据CPU的数量创建CPU*2数量的工作者线程。
    for I:=0 to LocalSI.dwNumberOfProcessors * 2 -1 do
    begin
      ServerWorkerThread := TServerWorkerThread.Create(CompletionPort,Self);
    end;
  end;
  result := FCompletionPort
end;

function TIOCPCustomComponent.GetHandle: THandle;
begin
  if FHandle=0 then
    FHandle := AllocateHWnd(WMProc);
  result := FHandle;
end;


function TIOCPCustomComponent.SendData(ASkt: TSocket; AData: PChar;
  ADataLen: Int64): Boolean;
var
  PerIoData: LPPER_IO_OPERATION_DATA ;
  SendBytes: DWORD;
  SendBuf:PChar;
begin
  //申请内存
  SendBuf := AData;
  SendBuf := GetMemory(ADataLen);
  StrCopy(SendBuf,AData);
  try
    //在这里申请一个发送数据的"单IO数据结构"
    PerIoData := LPPER_IO_OPERATION_DATA(GlobalAlloc(GPTR, sizeof(PER_IO_OPERATION_DATA)));
    if (PerIoData = nil) then
    begin
      Result:=false;
      exit;
    end;
    ZeroMemory(@PerIoData.Overlapped, sizeof(OVERLAPPED));

    //设置发送标记
    PerIoData.BytesRECV := 0;
    PerIoData.DataBuf.len := Min(DATA_BUFSIZE,ADataLen);
    PerIoData.DataBuf.buf:= SendBuf;
    PerIoData.BytesSEND := ADataLen;
    PerIoData.SendPointer := SendBuf;
    PerIoData.SendPointerSize := ADataLen;
    Flags := 0;
    //使用WSASend函数将数据发送
    if (WSASend(ASkt, @(PerIoData.DataBuf), 1, @SendBytes, 0,@(PerIoData.Overlapped), nil) = SOCKET_ERROR) then
    begin
      if (WSAGetLastError() <> ERROR_IO_PENDING) then
      begin
        WriteLog(Inttostr(WSAGetLastError()));
      //最近在检查代码的时候发现以前这里只是使用Exit来退出是不正确的。这里需要删除申请的单IO数据结构，否子会出现内存泄露。 (2008年3月24日)
      //Exit;
      //表示发送失败，以后也不会有处理在工作者线程处出现。
        if PerIoData <> nil then
        begin
          GlobalFree(DWORD(PerIoData));
        end;
        Result:=false;
        Exit;
      end;
    end;
    Result:=true;
  except
    Result:=false;
  end;

end;

procedure TIOCPCustomComponent.SetOnConnect(
  const Value: TIOCPConnectEvent);
begin
  FOnConnect := Value;
end;

procedure TIOCPCustomComponent.SetOnDisConnect(
  const Value: TIOCPDisConnectEvent);
begin
  FOnDisConnect := Value;
end;

procedure TIOCPCustomComponent.SetOnRead(const Value: TIOCPReadEvent);
begin
  FOnRead := Value;
end;

procedure TIOCPCustomComponent.WMProc(var Msg: TMessage);
begin

end;

procedure TIOCPCustomComponent.WriteLog(ALogText: string);
begin

end;

{ TIOCPServer }

function TIOCPServer.BroadCast(AData: PChar; ADataLen: Int64): Integer;
var
  i: integer;
begin
  for i := 0 to LinkCount -1 do
  begin
    SendData(StrToInt(FlinkList.Strings[i]),AData,ADataLen);
  end;
end;

constructor TIOCPServer.Create(AOwner: TComponent);
begin
  inherited Create(AOwner);
  //CompletionPort
  //MessageBox(Handle, PChar(Inttostr(CompletionPort)), '提示：', MB_OK);
  FLinkList := TStringList.Create;
  FPort := 6666;
  FActive := false;
end;

destructor TIOCPServer.Destroy;
begin
  if Assigned(FLinkList) then FreeAndNil(FLinkList);
  inherited;
end;

procedure TIOCPServer.DoConnect(ASkt: TSocket;ALinkInfo: PLinkInfo;var Accept: Boolean);
begin
  inherited;
  if Assigned(FLinkList) and Accept then
  begin
    FLinkList.AddObject(IntToStr(ASkt),TObject(ALinkInfo));
  end;
end;

procedure TIOCPServer.DoDisConnect(ASkt: TSocket);
begin
  inherited;
  if assigned(FLinkList) then
  begin
    FLinkList.Delete(FLinkList.IndexOf(IntToStr(ASkt)));
  end;
end;

function TIOCPServer.GetLinkCount: integer;
begin
  result := FLinkList.Count; 
end;

procedure TIOCPServer.SetActive(const Value: Boolean);
var
  sto:TSockAddrIn;
  Acceptsc :TSocket;
  Listensc: integer;
  Flags: integer;
  RecvBytes: DWord;
begin
  if FActive <> Value then FActive := Value;

  if FActive then
  begin
    if not Assigned(FThreadServer) then
    begin
      FThreadServer := TThreadServer.Create(self);
    end;
  end
  else
  begin

  end;
end;

procedure TIOCPServer.SetPort(const Value: Integer);
begin
  if FPort <> Value then
  begin
    if Value <= 0 then raise Exception.Create('无效端口号！');
    FPort := Value;
  end;
end;
{ TThreadServer }

constructor TThreadServer.Create(Owner: TIOCPServer);