DelphiIOCP研究笔记<七>===接收数据的解码器(Decoder)和数据处理

今天完成了第三点，初步按照netty 的做法制作了Decoder，由于我现在用的2007还没有泛型，所有我使用的返回TObject做法

我先介绍下netty的处理数据的流程

1.IOCP接收的数据。

2.写入到套接字对应的缓存。

3.调用Decoder，进行解码。

4.如果解码成功调用套接字对应的数据处理方法。

第一步和第二步我把他归纳到IOCP的底层通信要完成的功能。

第三步在这里需要和客户端定义协议。和数据打包和拆包的格式。

第四步处理具体的业务逻辑。

下面我一一贴出代码

第一步由IOCP完成。

第二步

if PerIoData.IO_TYPE = IO_TYPE_Recv then
begin
            //加入到套接字对应的缓存中，处理逻辑
            lvClientContext.RecvBuffer(PerIoData.DataBuf.buf, PerIoData.Overlapped.InternalHigh);
            TIODataMemPool.instance.giveBackIOData(PerIoData);

//加入cs控制，避免多工作线程，写入数据混乱

procedure TClientContext.RecvBuffer(buf: PAnsiChar; len: Cardinal);
var
  lvObject:TObject;
begin
  FCS.Enter;
  try
    //加入到套接字对应的缓存
    FBuffers.AddBuffer(buf, len);

    //调用注册的解码器<进行解码>
    lvObject := TContextFactory.instance.FDecoder.Decode(FBuffers);
    if lvObject <> nil then
    try
      //解码成功，调用业务逻辑的处理方法
      dataReceived(lvObject);
    finally
      lvObject.Free;
    end;
  finally
    FCS.Leave;
  end;
end;

//在运行IOCP服务时注册解码器

procedure TfrmMain.btnIOCPAPIRunClick(Sender: TObject);
var
  lvData:PInteger;

  lvdwThreadId:DWORD;
begin
  __ListenPort := StrToInt(edtPort.Text);

  //注册解码器
  TContextFactory.instance.registerDecoder(FDecoder);
    
  CreateThread(nil, 0, @D10_IOCPRun, Pointer(__ListenPort), 0, lvdwThreadId);
end;

JSonStream的解码器<可以根据自己的格式进行编写解码器>

function TJSonStreamDecoder.Decode(const inBuf: TBufferLink): TObject;
var
  lvJSonLength, lvStreamLength:Integer;
  lvData:String;
  lvBuffer:array of Char;
  lvBufData:PAnsiChar;
  lvStream:TMemoryStream;
  lvJsonStreamObject:TJSonStreamObject;
begin
  Result := nil;

  //如果缓存中的数据长度不够包头长度，解码失败<json字符串长度,流长度>
  if (inBuf.validCount < SizeOf(Integer) + SizeOf(Integer)) then
  begin
    Exit;
  end;

  //记录读取位置
  inBuf.markReaderIndex;
  inBuf.readBuffer(@lvJSonLength, SizeOf(Integer));
  inBuf.readBuffer(@lvStreamLength, SizeOf(Integer));

  //如果缓存中的数据不够json的长度和流长度<说明数据还没有收取完毕>解码失败
  if inBuf.validCount < (lvJSonLength + lvStreamLength) then
  begin
    //返回buf的读取位置
    inBuf.restoreReaderIndex;
    exit;
  end;


  //解码成功
  lvJsonStreamObject := TJSonStreamObject.Create;
  Result := lvJsonStreamObject;

  //读取json字符串
  if lvJSonLength > 0 then
  begin
    SetLength(lvData, lvJSonLength);
    ZeroMemory(@lvData[1], lvJSonLength);
    inBuf.readBuffer(@lvData[1], lvJSonLength);
    lvJsonStreamObject.JSon := SO(lvData);
  end;


  //读取流数据 
  if lvStreamLength > 0 then
  begin
    lvStream := TMemoryStream.Create;
    lvStream.Size := lvStreamLength;
    inBuf.readBuffer(lvStream.Memory, lvStreamLength);
    lvJsonStreamObject.setStream(lvStream);
  end;

end;

//业务逻辑处理

//这里我简单的做了一个客户端发送文件的列子。

procedure TClientContext.dataReceived(const pvDataObject:TObject);
var
  lvJsonStream:TJSonStreamObject;
  lvFile:String;
begin
  lvJsonStream := TJSonStreamObject(pvDataObject);

  //客户端发送文件
  if lvJsonStream.JSon.I['cmdIndex'] = 102 then
  begin
    lvFile := ExtractFilePath(ParamStr(0)) + 'tempFile\';
    ForceDirectories(lvFile);
    lvFile := lvFile + lvJsonStream.JSon.S['file'];
    TMemoryStream(lvJsonStream.Stream).Position := 0;
    TMemoryStream(lvJsonStream.Stream).SaveToFile(lvFile);
  end;


  TLogClientWrapper.logINfo(lvJsonStream.JSon.AsJSon(True));

end;

代码比较简单

我还是把我的demo贴出来<还是放在360云盘上面>

Delphi-IOCP源码下载