从 OnRead 事件传递到分离的线程

Question

我在一个项目中工作，该项目需要接收实时网络摄像头的连续帧，我发现在我的测试中效果很好。现在想知道如何在类似于服务器 multiclient/multithread 的方法的 TThread (Socket NonBlocking) 中进行接收？我试过了，但是服务器没有收到来自客户端的 none 帧。我希望你能帮助我。

服务器:

uses
  System.Win.ScktComp, Winapi.WinSock, Vcl.Imaging.jpeg, System.Math;

type
  TMyThread = class(TThread)
  private
    Socket: TCustomWinSocket;
  protected
    procedure Execute; override;
  public
    constructor Create(aSocket: TCustomWinSocket);
  end;

type
  TForm1 = class(TForm)
    Button1: TButton;
    Image1: TImage;
    ServerSocket1: TServerSocket;
    procedure ServerSocket1Accept(Sender: TObject; Socket: TCustomWinSocket);
    procedure ServerSocket1ClientConnect(Sender: TObject; Socket: TCustomWinSocket);
    procedure ServerSocket1ClientDisconnect(Sender: TObject; Socket: TCustomWinSocket);
    procedure ServerSocket1ClientError(Sender: TObject; Socket: TCustomWinSocket; ErrorEvent: TErrorEvent; var ErrorCode: Integer);
    procedure Button1Click(Sender: TObject);
    procedure ServerSocket1Listen(Sender: TObject; Socket: TCustomWinSocket);
  private
    { Private declarations }
    MyThread: TMyThread;
  public
    { Public declarations }
  end;

var
  Form1: TForm1;

implementation

{$R *.dfm}

constructor TMyThread.Create(aSocket: TCustomWinSocket);
begin
  inherited Create(True);
  Socket := aSocket;
  FreeOnTerminate := True;
end;

procedure TForm1.Button1Click(Sender: TObject);
begin
  ServerSocket1.Port := 1234;
  ServerSocket1.Active := true;
end;

procedure TForm1.ServerSocket1Accept(Sender: TObject; Socket: TCustomWinSocket);
begin
  MyThread := TMyThread.Create(Socket);
  MyThread.Start;
end;

procedure TForm1.ServerSocket1ClientConnect(Sender: TObject; Socket: TCustomWinSocket);
begin
  Socket.Data := nil;
end;

procedure TForm1.ServerSocket1ClientDisconnect(Sender: TObject; Socket: TCustomWinSocket);
begin
  if Socket.Data <> nil then
    TMemoryStream(Socket.Data).Free;
end;

procedure TForm1.ServerSocket1ClientError(Sender: TObject; Socket: TCustomWinSocket; ErrorEvent: TErrorEvent; var ErrorCode: Integer);
begin
  ErrorCode := 0;
end;

procedure TForm1.ServerSocket1Listen(Sender: TObject; Socket: TCustomWinSocket);
begin
  ShowMessage('Server listen on port: ' + IntToStr(Socket.LocalPort));
end;

procedure TMyThread.Execute;
var
  Stream: TMemoryStream;
  BytesReceived: Integer;
  StreamSize, TempSize: Int32;
  BytesRemaining: Int64;
  P: PByte;
  ChunkSize: Integer;
  jpg: TJpegImage;
const
  MaxChunkSize: Int64 = 8192;
begin
  while Socket.Connected do
  begin
    Stream := TMemoryStream(Socket.Data);

    if Stream = nil then
    begin
      if Socket.ReceiveLength < SizeOf(TempSize) then
        Exit;
      BytesReceived := Socket.ReceiveBuf(TempSize, SizeOf(TempSize));
      if BytesReceived <= 0 then
        Exit;
      StreamSize := ntohl(TempSize);
      Stream := TMemoryStream.Create;
      Socket.Data := Stream;
      Stream.Size := StreamSize;
      BytesRemaining := StreamSize;
    end
    else
      BytesRemaining := Stream.Size - Stream.Position;

    if BytesRemaining > 0 then
    begin
      P := PByte(Stream.Memory);
      if Stream.Position > 0 then
        Inc(P, Stream.Position);
      repeat
        ChunkSize := Integer(Min(BytesRemaining, MaxChunkSize));
        BytesReceived := Socket.ReceiveBuf(P^, ChunkSize);
        if BytesReceived <= 0 then
          Exit;
        Inc(P, BytesReceived);
        Dec(BytesRemaining, BytesReceived);
        Stream.Seek(BytesReceived, soCurrent);
      until BytesRemaining = 0;
    end;

    try
      jpg := TJpegImage.Create;
      try
        Stream.Position := 0;
        jpg.LoadFromStream(Stream);
        Synchronize(
          procedure
          begin
            Form1.Image1.Picture.Assign(jpg);
          end);
      finally
        jpg.Free;
      end;
    finally
      Socket.Data := nil;
      Stream.Free;
    end;
  end;
end;

end.

Answer 1

您需要在 线程阻塞模式 中使用 TServerSocket 才能有效地将工作线程与其接受的客户端一起使用。非阻塞模式和工作线程不能很好地混合在一起。

非阻塞模式的发明是为了能够在主 UI 线程中使用 TClientSocket 和 TServerSocket 而不会阻塞它。但是当在主 UI 线程之外使用套接字时，非阻塞模式的用处很少（只是一些不适用于您的情况的极端情况）。在内部，TCustomWinSocket分配一个HWND来检测套接字activity，当用于非阻塞时，HWND需要一个消息循环。但是由于每个接受的客户端套接字都是在您的工作线程之外创建的，因此它们的 HWND 将无法由您在线程中运行的任何消息循环提供服务。因此，无论如何您都需要使用线程阻塞模式。

此外，无论如何，使用线程阻塞模式将大大简化您的套接字 I/O 代码。

试试像这样的东西：

unit Unit1;

interface

uses
  ..., System.Win.ScktComp;

type
  TForm1 = class(TForm)
    Button1: TButton;
    Image1: TImage;
    ServerSocket1: TServerSocket;
    procedure Button1Click(Sender: TObject);
    procedure ServerSocket1ClientError(Sender: TObject; Socket: TCustomWinSocket; ErrorEvent: TErrorEvent; var ErrorCode: Integer);
    procedure ServerSocket1GetThread(Sender: TObject; ClientSocket: TServerClientWinSocket; var SocketThread: TServerClientThread);
    procedure ServerSocket1Listen(Sender: TObject; Socket: TCustomWinSocket);
  private
    { Private declarations }
  public
    { Public declarations }
  end;

var
  Form1: TForm1;

implementation

uses
  Winapi.WinSock, Vcl.Imaging.jpeg, System.Math;

{$R *.dfm}

type
  TMyThread = class(TServerClientThread)
  protected
    procedure ClientExecute; override;
  end;

procedure TForm1.Button1Click(Sender: TObject);
begin
  // this can be set at design-time, if desired...
  ServerSocket1.ServerType := TServerType.stThreadBlocking;

  // so can this...
  ServerSocket1.Port := 1234;

  ServerSocket1.Active := True;
end;

procedure TForm1.ServerSocket1ClientError(Sender: TObject; Socket: TCustomWinSocket; ErrorEvent: TErrorEvent; var ErrorCode: Integer);
begin
  ErrorCode := 0;
end;

procedure TForm1.ServerSocket1GetThread(Sender: TObject; ClientSocket: TServerClientWinSocket; var SocketThread: TServerClientThread);
begin
  SocketThread := TMyThread.Create(False, ClientSocket);
end;

procedure TForm1.ServerSocket1Listen(Sender: TObject; Socket: TCustomWinSocket);
begin
  ShowMessage('Server listen on port: ' + IntToStr(Socket.LocalPort));
end;

procedure TMyThread.ClientExecute;
var
  Stream: TMemoryStream;
  StreamSize: Int32;
  jpg: TJpegImage;

  function DoRead(Buffer: Pointer; BufSize: Int64): Boolean;
  const
    MaxChunkSize: Int64 = 8192;
  var
    P: PByte;
    BytesReceived: Integer;
    ChunkSize: Integer;
  begin
    Result := False;
    P := PByte(Buffer);
    while BufSize > 0 do
    begin
      ChunkSize := Integer(Min(BufSize, MaxChunkSize));
      BytesReceived := ClientSocket.ReceiveBuf(P^, ChunkSize);
      if BytesReceived <= 0 then
        Exit;
      Inc(P, BytesReceived);
      Dec(BufSize, BytesReceived);
    end;
    Result := True;
  end;

begin
  while (not Terminated) and ClientSocket.Connected do
  begin
    if not DoRead(@StreamSize, SizeOf(StreamSize)) then Exit;
    StreamSize := ntohl(StreamSize);
    if StreamSize <= 0 then Continue;
    jpg := TJpegImage.Create;
    try
      Stream := TMemoryStream.Create;
      try
        Stream.Size := StreamSize;
        if not DoRead(Stream.Memory, StreamSize) then Exit;
        Stream.Position := 0;
        jpg.LoadFromStream(Stream);
      finally
        Stream.Free;
      end;
      Synchronize(
        procedure
        begin
          Form1.Image1.Picture.Assign(jpg);
        end
      );
    finally
      jpg.Free;
    end;
  end;
end;

end.

也就是说，我强烈建议您停止使用 Borland 遗留的这些过时和弃用的 套接字组件。例如，Indy 10 预装在 IDE 中，并且有一个 TIdTCPServer 组件，这将进一步大大简化上述线程逻辑（TIdTCPServer 是一个多线程组件，将为您管理每个客户端线程），例如：

unit Unit1;

interface

uses
  ..., IdContext, IdTCPServer;

type
  TForm1 = class(TForm)
    Button1: TButton;
    Image1: TImage;
    IdTCPServer1: TIdTCPServer;
    procedure Button1Click(Sender: TObject);
    procedure IdTCPServer1Connect(AContext: TIdContext);
    procedure IdTCPServer1Execute(AContext: TIdContext);
  private
    { Private declarations }
  public
    { Public declarations }
  end;

var
  Form1: TForm1;

implementation

uses
  Vcl.Imaging.jpeg, System.Math;

{$R *.dfm}

procedure TForm1.Button1Click(Sender: TObject);
begin
  IdTCPServer1.DefaultPort := 1234;
  IdTCPServer1.Active := True;
  ShowMessage('Server listen on port: ' + IntToStr(IdTCPServer1.DefaultPort));
end;

procedure TForm1.IdTCPServer1Connect(AContext: TIdContext);
begin
  // tell ReadStream() to read the stream size as an Int32 and not as an Int64...
  AContext.Connection.IOHandler.LargeStream := False;
end;

procedure TForm1.IdTCPServer1Execute(AContext: TIdContext);
var
  Stream: TMemoryStream;
  jpg: TJpegImage;
begin
  // OnExecute is a looped event, it is called in a continuous
  // loop for the lifetime of the TCP connection...

  jpg := TJpegImage.Create;
  try
    Stream := TMemoryStream.Create;
    try
      // ReadStream() can read the stream size first, then read the stream data...
      AContext.Connection.IOHandler.ReadStream(Stream, -1, False);

      Stream.Position := 0;
      jpg.LoadFromStream(Stream);
    finally
      Stream.Free;
    end;
    TThread.Synchronize(nil,
      procedure
      begin
        Form1.Image1.Picture.Assign(jpg);
      end
    );
  finally
    jpg.Free;
  end;
end;

end.

从 OnRead 事件传递到分离的线程

Pass from OnRead event to separated Thread

multithreading

sockets

delphi

client-server

delphi-10.3-rio