问如何在TIdTcpServer和TIdTCPClient (如洪水)之间快速交换数据
EN

Stack Overflow用户

提问于 2020-12-29 09:31:07

回答 1查看 349关注 0票数 4

我有一个简单的TidTCPServer在控制台上工作并接受数据。我的问题是当客户端发送流，但是有很高的速度交换数据时，服务器在70行之后冻结，服务器的CPU负载达到70%；我不知道如何解决，而不增加每次发送之间的睡眠。下面是客户机和服务器的示例。你能帮我解决这个(服务器端)谢谢。

program Srv;

{$I Synopse.inc}

uses
  {$IFDEF UNIX}{$IFDEF UseCThreads}
  cthreads,
  {$ENDIF}{$ENDIF}
  Classes, SysUtils, CustApp, Generics.Collections, IdTCPServer, IdCustomTCPServer, IdContext, IdGlobal, Db, mORMot, mORMotSQLite3, IdSync, functions, SynCommons, SynSQLite3Static;

type

  { TMyApplication }
  TMyApplication = class(TCustomApplication)

   var IdTCPServer: TIdTCPServer;

   protected
    procedure DoRun; override;
    procedure ServerOnConnect(AContext: TIdContext);
    procedure ServerOnExecute(AContext: TIdContext);
    function ReceiveStream(AContext: TIdContext;Size:integer; var AStream: TStream);

  public
    constructor Create(TheOwner: TComponent); override;
    destructor Destroy; override;
  end;

  type
    TLog = class(TIdNotify)
    protected
      FMsg: string;
      procedure DoNotify; override;
    public
      class procedure LogMsg(const AMsg: string);
    end;

{ TMyApplication }

    procedure TLog.DoNotify;
    var i:integer;
    begin
     writeln(FMsg);
    end;

    class procedure TLog.LogMsg(const AMsg: string);
    begin
      with TLog.Create do
      try
        FMsg := AMsg;
        Notify;
      except
        Free;
        raise;
      end;
    end;

function TMyApplication.ReceiveStream(AContext: TIdContext; var AStream: TStream)
  : Boolean; overload;
var
  LSize: LongInt;
begin
  Result := True;
  try
    LSize := AContext.Connection.IOHandler.ReadLongInt();
    AContext.Connection.IOHandler.ReadStream(AStream,LSize, False)
    AStream.Seek(0,soFromBeginning);
  except
    Result := False;
  end;
end;      

procedure TMyApplication.ServerOnExecute(AContext: TIdContext);
var AStream:TMemoryStream;
begin

if (Acontext.Connection.IOHandler.InputBufferIsEmpty) then
  begin
    TLog.LogMsg('--: '+random(100000).ToString); //After executing Client this line is displayed 70 time and CPU load is from 40 % to 70%
    AStream:=TMemoryStream.Create;
    try 
      ReceiveStream(AContext,TStream(AStream)); 
      // .. here we use AStream to execute some stuff  
    finally
      Astream.free;
    end;        
end;

procedure TMyApplication.ServerOnConnect(AContext: TIdContext);
begin
 TLog.LogMsg('connect');        
end;

procedure TMyApplication.DoRun;
begin

    IdTCPServer := tIdTCPServer.Create;
    IdTCPServer.ListenQueue := 15;
    IdTCPServer.MaxConnections := 0;
    IdTCPServer.TerminateWaitTime := 5000;
    with IdTCPServer.Bindings.Add
    do begin
      IP   := '0.0.0.0';
      Port := 80;
      IPVersion:=Id_IPv4;
    end;
    IdTCPServer.OnConnect := ServerOnConnect;
    IdTCPServer.OnDisconnect := ServerOnDiconnect;
    IdTCPServer.OnExecute := ServerOnExecute;
    IdTCPServer.Active := True;

  while true do
   begin
    Classes.CheckSynchronize() ;
    sleep(10);
   end;

  readln;        

  Terminate;
end;

constructor TMyApplication.Create(TheOwner: TComponent);
begin

  inherited Create(TheOwner);
  StopOnException := True;

end;


destructor TMyApplication.Destroy;
begin
  IdTCPServer.Free;
  inherited Destroy;

end;

var
  Application: TMyApplication;
begin
  Application := TMyApplication.Create(nil);
  Application.Title := 'My Application';
  Application.Run;
  Application.Free;
end.

客户端

function TForm1.SendStream(AClient: TIdTCPClient; AStream: TStream): Boolean; overload;
var
  StreamSize: LongInt;
begin
  try
    Result := True;
    try
      AStream.Seek(0,soFromBeginning);
      StreamSize := (AStream.Size);
      AClient.IOHandler.Write(LongInt(StreamSize));
      AClient.IOHandler.WriteBufferOpen;
      AClient.IOHandler.Write(AStream, 0, False);
      AClient.IOHandler.WriteBufferFlush;
    finally
      AClient.IOHandler.WriteBufferClose;
    end;
  except
    Result := False;
  end;
end;   
    
procedure TForm1.Button1Click(Sender: TObject);
var 
  Packet:TPacket;
  AStream:TMemoryStream;
begin
for i:=0 to 1000 do 
  begin
    Application.ProcessMessages;
    With Packet do
                 begin
                   MX               := random(10000);
                   MY               := random(10000);
                 end;
     AStream:=TMemoryStream.Create;
     try
        AStream.Write(Packet,SizeOf(TPacket));
        SendStream(IdTCPClientCmd,TStream(AStream));
     finally
        AStream.Free;
     end;
  end;

end;

indy10

delphi

delphi-7

indy

lazarus

回答 1

Stack Overflow用户

发布于 2020-12-29 10:26:55

在服务器端，您的InputBufferIsEmpty()检查是向后的。如果客户端发送了大量数据，InputBufferIsEmpty()可能最终会变成False，这将导致服务器代码进入一个严格的、不需要读取任何内容的循环。只需完全摆脱检查，让ReceiveStream()阻塞，直到有可用的数据包可阅读。

另外，为什么要将服务器的ListenQueue设置为15，而将MaxConnections设置为0呢？MaxConnections=0将迫使服务器立即关闭每个已被接受的客户端连接，因此OnExecute事件将永远不会被调用。

在客户端，没有必要在每次循环迭代中销毁和重新创建TMemoryStream，您应该重用该对象。

但是更重要的是，您没有正确地使用写缓冲，所以要么修复它，要么摆脱它。我会做后者，因为您正在发送大量的小数据包，所以让TCP的默认合并处理缓冲。

TIdIOHandler.Write(TStream)/TIdIOHandler.ReadStream()可以为您交换流大小，不需要手动执行。

试一试：

服务器

program Srv;

{$I Synopse.inc}

uses
  {$IFDEF UNIX}{$IFDEF UseCThreads}
  cthreads,
  {$ENDIF}{$ENDIF}
  Classes, SysUtils, CustApp, Generics.Collections, IdTCPServer, IdCustomTCPServer, IdContext, IdGlobal, Db, mORMot, mORMotSQLite3, IdSync, functions, SynCommons, SynSQLite3Static;

type

  { TMyApplication }
  TMyApplication = class(TCustomApplication)
  var
    IdTCPServer: TIdTCPServer;
  protected
    procedure DoRun; override;
    procedure ServerOnConnect(AContext: TIdContext);
    procedure ServerOnExecute(AContext: TIdContext);
    function ReceiveStream(AContext: TIdContext; Size: Integer; var AStream: TStream);
  public
    constructor Create(TheOwner: TComponent); override;
    destructor Destroy; override;
  end;

  type
    TLog = class(TIdNotify)
    protected
      FMsg: string;
      procedure DoNotify; override;
    public
      class procedure LogMsg(const AMsg: string);
    end;

{ TMyApplication }

procedure TLog.DoNotify;
begin
  WriteLn(FMsg);
end;

class procedure TLog.LogMsg(const AMsg: string);
begin
  with TLog.Create do
  try
    FMsg := AMsg;
    Notify;
  except
    Free;
    raise;
  end;
end;

function TMyApplication.ReceiveStream(AContext: TIdContext; AStream: TStream): Boolean; overload;
begin
  try
    AContext.Connection.IOHandler.ReadStream(AStream, -1, False);
    AStream.Position := 0;
    Result := True;
  except
    Result := False;
  end;
end;      

procedure TMyApplication.ServerOnExecute(AContext: TIdContext);
var
  AStream: TMemoryStream;
begin
  AStream := TMemoryStream.Create;
  try 
    if not ReceiveStream(AContext, AStream) then
    begin
      AContext.Connection.Disconnect;
      Exit;
    end;
    TLog.LogMsg('--: '+random(100000).ToString); //After executing Client this line is displayed 70 time and CPU load is from 40 % to 70%
    // .. here we use AStream to execute some stuff  
  finally
    AStream.Free;
  end;        
end;

procedure TMyApplication.ServerOnConnect(AContext: TIdContext);
begin
  TLog.LogMsg('connect');        
  AContext.Connection.IOHandler.LargeStream := False;
end;

procedure TMyApplication.DoRun;
begin
  IdTCPServer := TIdTCPServer.Create;
  IdTCPServer.ListenQueue := 15;
  IdTCPServer.MaxConnections := 1;
  IdTCPServer.TerminateWaitTime := 5000;
  with IdTCPServer.Bindings.Add do
  begin
    IP   := '0.0.0.0';
    Port := 80;
    IPVersion := Id_IPv4;
  end;
  IdTCPServer.OnConnect := ServerOnConnect;
  IdTCPServer.OnDisconnect := ServerOnDiconnect;
  IdTCPServer.OnExecute := ServerOnExecute;
  IdTCPServer.Active := True;

  while True do
  begin
    Classes.CheckSynchronize();
    Sleep(10);
  end;

  ReadLn;
  Terminate;
end;

constructor TMyApplication.Create(TheOwner: TComponent);
begin
  inherited Create(TheOwner);
  StopOnException := True;
end;

destructor TMyApplication.Destroy;
begin
  IdTCPServer.Free;
  inherited Destroy;
end;

var
  Application: TMyApplication;
begin
  Application := TMyApplication.Create(nil);
  Application.Title := 'My Application';
  Application.Run;
  Application.Free;
end.

客户端

function TForm1.SendStream(AClient: TIdTCPClient; AStream: TStream): Boolean; overload;
begin
  try
    AClient.IOHandler.LargeStream := False; // <-- or, set this 1 time after TIdTCPClient.Connect() exits...
    AClient.IOHandler.Write(AStream, 0, True);
    Result := True;
  except
    Result := False;
  end;
end;   
    
procedure TForm1.Button1Click(Sender: TObject);
var 
  Packet: TPacket;
  AStream: TMemoryStream;
  i: Integer;
begin
  AStream := TMemoryStream.Create;
  try
    AStream.Size := SizeOf(TPacket);
    for i := 0 to 1000 do 
    begin
      Application.ProcessMessages;
      with Packet do
      begin
        MX := random(10000);
        MY := random(10000);
      end;
      AStream.Position := 0;
      AStream.Write(Packet, SizeOf(TPacket));
      SendStream(IdTCPClientCmd, AStream);
    end;
  finally
    AStream.Free;
  end;
end;