在阻塞主线程中使用 TEvent 和 MsgWaitForMultipleObjects

Question

我发现了这个 Remy 的有趣代码。

是否可以这样做，以便我可以执行多个线程并等待它们全部完成，然后继续主线程？我这样试过，但没有成功...

procedure Requery(DataList: TStringList);
var
  Event: TEvent;
  H: THandle;
  OpResult: array of Boolean;
  i: Integer;
begin
  Event := TEvent.Create;
  try
    SetLength(OpResult, DataList.Count); 
    for i:=0 to DataList.Count-1 do begin
      TThread.CreateAnonymousThread(
        procedure
        begin
          try
            // run query in thread
            OpResult[i]:=IsMyValueOK(DataList.Strings[i]);
          finally
            Event.SetEvent;
          end;
        end
      ).Start;
      H := Event.Handle;
    end;
    while MsgWaitForMultipleObjects(1, H, False, INFINITE, QS_ALLINPUT) = (WAIT_OBJECT_0+1) do Application.ProcessMessages;
    
    for i:=Low(OpResult) to High(OpResult) do begin
      Memo1.Lines.Add('Value is: ' + BoolToStr(OpResult[i], True));
    end;
  finally
    Event.Free;
  end;

  // Do next jobs with query
  ...
end;

Answer 1

Can this be done so I can do multiple threads and wait until they are all finished

是的。您只需要创建多个 TEvent 对象，每个对象对应 TThread，然后将它们的所有 Handle 存储在一个数组中以传递给 MsgWaitForMultipleObjects():

procedure Requery(DataList: TStringList);
var
  Events: array of TEvent;
  H: array of THandle;
  OpResult: array of Boolean;
  i: Integer;
  Ret, Count: DWORD;

  // moved into a helper function so that the anonymous procedure
  // can capture the correct Index...
  procedure StartThread(Index: integer);
  begin
    Events[Index] := TEvent.Create;
    TThread.CreateAnonymousThread(
      procedure
      begin
        try
          // run query in thread
          OpResult[Index] := IsMyValueOK(DataList.Strings[Index]);
        finally
          Events[Index].SetEvent;
        end;
      end
    ).Start;
    H[Index] := Events[Index].Handle;
  end;

begin
  if DataList.Count > 0 then
  begin
    SetLength(Events, DataList.Count);
    SetLength(H, DataList.Count);
    SetLength(OpResult, DataList.Count);

    try
      for i := 0 to DataList.Count-1 do begin
        StartThread(i);
      end;

      Count := Length(H);
      repeat
        Ret := MsgWaitForMultipleObjects(Count, H[0], False, INFINITE, QS_ALLINPUT);
        if Ret = WAIT_FAILED then RaiseLastOSError;
        if Ret = (WAIT_OBJECT_0+Count) then
        begin
          Application.ProcessMessages;
          Continue;
        end;
        for i := Integer(Ret-WAIT_OBJECT_0)+1 to High(H) do begin
          H[i-1] := H[i];
        end;
        Dec(Count);
      until Count = 0;

      for i := Low(OpResult) to High(OpResult) do begin
        Memo1.Lines.Add('Value is: ' + BoolToStr(OpResult[i], True));
      end;
    finally
      for i := Low(Events) to High(Events) do begin
        Events[i].Free;
      end;
    end;
  end;

  // Do next jobs with query
  ...
end;

也就是说，您也可以摆脱 TEvent 对象并等待 TThread.Handle 对象。当线程完全终止时，线程的 Handle 发出等待操作的信号。唯一的问题是 TThread.CreateAnonymousThread() 创建了一个 TThread，其 FreeOnTerminate 属性是 True，因此您必须手动将其关闭：

procedure Requery(DataList: TStringList);
var
  Threads: array of TThread;
  H: array of THandle;
  OpResult: array of Boolean;
  i: Integer;
  Ret, Count: DWORD;

  // moved into a helper function so that the anonymous procedure
  // can capture the correct Index...
  procedure StartThread(Index: integer);
  begin
    Threads[Index] := TThread.CreateAnonymousThread(
      procedure
      begin
        // run query in thread
        OpResult[Index] := IsMyValueOK(DataList.Strings[Index]);
      end
    );
    Threads[Index].FreeOnTerminate := False;
    H[Index] := Threads[Index].Handle;
    Threads[Index].Start;
  end;

begin
  try
    SetLength(Threads, DataList.Count);
    SetLength(H, DataList.Count);
    SetLength(OpResult, DataList.Count);

    for i := 0 to DataList.Count-1 do begin
      StartThread(i);
    end;

    Count := Length(H);
    repeat
      Ret := MsgWaitForMultipleObjects(Count, H[0], False, INFINITE, QS_ALLINPUT);
      if Ret = WAIT_FAILED then RaiseLastOSError;
      if Ret = (WAIT_OBJECT_0+Count) then
      begin
        Application.ProcessMessages;
        Continue;
      end;
      for i := Integer(Ret-WAIT_OBJECT_0)+1 to High(H) do begin
        H[i-1] := H[i];
      end;
      Dec(Count);
    until Count = 0;

    for i := Low(OpResult) to High(OpResult) do begin
      Memo1.Lines.Add('Value is: ' + BoolToStr(OpResult[i], True));
    end;
  finally
    for i := Low(Threads) to High(Threads) do begin
      Threads[i].Free;
    end;
  end;

  // Do next jobs with query
  ...
end;

无论哪种方式，请注意 MsgWaitForMultipleObjects() 被限制为一次最多等待 63 (MAXIMUM_WAIT_OBJECTS[64] - 1) 个句柄。 WaitForMultipleObjects() 文档解释了如何解决该限制，如果您需要：

To wait on more than MAXIMUM_WAIT_OBJECTS handles, use one of the following methods:

Create a thread to wait on MAXIMUM_WAIT_OBJECTS handles, then wait on that thread plus the other handles. Use this technique to break the handles into groups of MAXIMUM_WAIT_OBJECTS.

Call RegisterWaitForSingleObject to wait on each handle. A wait thread from the thread pool waits on MAXIMUM_WAIT_OBJECTS registered objects and assigns a worker thread after the object is signaled or the time-out interval expires.

或者，您可以简单地以较小的批次处理列表，比如一次不超过 50-60 项。

在阻塞主线程中使用 TEvent 和 MsgWaitForMultipleObjects

Using TEvent and MsgWaitForMultipleObjects in blocking main thread

multithreading

delphi

events

blocking

tthread