如何可靠地等待刚刚创建的线程？

Question

考虑以下程序:

program TThreadBug;
{$APPTYPE CONSOLE}

uses
  SysUtils, Classes, Windows;

type
  TMyThread = class(TThread)
  protected
    procedure Execute; override;
  end;

procedure TMyThread.Execute;
var
  i: Integer;
begin
  for i := 1 to 5 do begin
    Writeln(i);
    Sleep(100);
  end;
end;

procedure UseTThread;
var
  Thread: TMyThread;
begin
  Writeln('TThread');
  Thread := TMyThread.Create;
  Thread.Terminate;
  Thread.WaitFor;
  Thread.Free;
  Writeln('Finished');
  Writeln;
end;

procedure UseTThreadWithSleep;
var
  Thread: TMyThread;
begin
  Writeln('TThreadWithSleep');
  Thread := TMyThread.Create;
  Sleep(100);
  Thread.Terminate;
  Thread.WaitFor;
  Thread.Free;
  Writeln('Finished');
  Writeln;
end;

begin
  UseTThread;
  UseTThreadWithSleep;
  Readln;
end.

输出是:

TThread
Finished

TThreadWithSleep
1
2
3
4
5
Finished

所以看起来,由于某种原因,主线程必须等待一段任意的时间才能终止并等待工作线程.我是否正确地认为这是一个错误TThread？有什么方法可以解决这个问题吗？我希望如果我让我的线程发出信号表明它已经启动(使用事件),那么这将解决问题.但这让我觉得很脏.

Answer 1

您可以将其称为错误或TThread设计缺陷,问题已被多次讨论过.例如参见http://sergworks.wordpress.com/2011/06/25/sleep-sort-and-tthread-corner-case/

问题是如果TThread.Terminated设置了标志太早,TThread.Execute则永远不会调用方法.所以你的情况就是不叫TThread.Terminate之前TThread.WaitFor.

Answer 2

我认为这种情况发生的原因已经被Serg的答案充分回答了,但我认为你通常不应该调用Thread.Terminate.调用它的唯一原因,如果您希望线程终止,例如在应用程序关闭时.如果您只想等到它完成,您可以调用WaitFor(或WaitForSingleObject).这是可能的,因为线程的句柄已经在其构造函数中创建,因此您可以立即调用它.

另外,我在这些线程上将FreeOnTerminate设置为true.让他们自己跑步和解脱.如果我想要通知它们,我可以使用WaitFor或OnTerminate事件.

这里只是一堆工作线程以阻塞方式清空队列的示例.

我认为你不应该需要这个,大卫,但也许别人可能会对一个例子感到满意.另一方面,你可能没有问这个问题只是为了改变对TThread执行不力的咆哮,对吧？;-)

首先是Queue类.我想,这不是一个真正的传统队列.在实际的多线程队列中,您应该能够在任何时候添加到队列中,即使处理处于活动状态也是如此.此队列要求您预先填充其项目,然后调用-blocking-run方法.此外,处理的项目将保存回队列.

type
  TQueue = class
  strict private
    FNextItem: Integer;
    FRunningThreads: Integer;
    FLock: TCriticalSection;
    FItems: TStrings; // Property...
  private

    // Signal from the thread that it is started or stopped.
    // Used just for indication, no real functionality depends on this.
    procedure ThreadStarted;
    procedure ThreadEnded;

    // Pull the next item from the queue.
    function Pull(out Item: Integer; out Value: string): Boolean;

    // Save the modified value back in the queue.
    procedure Save(Item: Integer; Value: string);

  public
    property Items: TStrings read FItems;
    constructor Create;
    destructor Destroy; override;

    // Process the queue. Blocking: Doesn't return until every item in the
    // queue is processed.
    procedure Run(ThreadCount: Integer);

    // Statistics for polling.
    property Item: Integer read FNextItem;
    property RunningThreads: Integer read FRunningThreads;
  end;

然后是Consumer线程.那一个很简单明了.它只是对队列的引用,以及在队列为空之前运行的execute方法.

  TConsumer = class(TThread)
  strict private
    FQueue: TQueue;
  protected
    procedure Execute; override;
  public
    constructor Create(AQueue: TQueue);
  end;

在这里,您可以看到这个模糊的"队列"的实现.它的主要方法是Pull和Save,消费者使用它来提取下一个项目,然后保存处理后的值.

另一个重要的方法是Run,它启动给定数量的工作线程并等待所有工作线程完成.所以这实际上是一个阻塞方法,它只在队列清空后返回.我在这里使用WaitForMultipleObjects,它允许您在需要添加额外技巧之前等待多达64个线程.它与您在问题中的代码中使用WaitForSingleObject相同.

看看Thread.Terminate从未被调用过？

{ TQueue }

constructor TQueue.Create;
// Context: Main thread
begin
  FItems := TStringList.Create;
  FLock := TCriticalSection.Create;
end;

destructor TQueue.Destroy;
// Context: Main thread
begin
  FLock.Free;
  FItems.Free;
  inherited;
end;

function TQueue.Pull(out Item: Integer; out Value: string): Boolean;
// Context: Consumer thread
begin
  FLock.Acquire;
  try
    Result := FNextItem < FItems.Count;
    if Result then
    begin
      Item := FNextItem;
      Inc(FNextItem);
      Value := FItems[Item];
    end;
  finally
    FLock.Release;
  end;
end;

procedure TQueue.Save(Item: Integer; Value: string);
// Context: Consumer thread
begin
  FLock.Acquire;
  try
    FItems[Item] := Value;
  finally
    FLock.Release;
  end;
end;

procedure TQueue.Run(ThreadCount: Integer);
// Context: Calling thread (TQueueBackgroundThread, or can be main thread)
var
  i: Integer;
  Threads: TWOHandleArray;
begin
  if ThreadCount <= 0 then
    raise Exception.Create('You no make sense no');
  if ThreadCount > MAXIMUM_WAIT_OBJECTS then
    raise Exception.CreateFmt('Max number of threads: %d', [MAXIMUM_WAIT_OBJECTS]);

  for i := 0 to ThreadCount - 1 do
    Threads[i] := TConsumer.Create(Self).Handle;

  WaitForMultipleObjects(ThreadCount, @Threads, True, INFINITE);
end;

procedure TQueue.ThreadEnded;
begin
  InterlockedDecrement(FRunningThreads);
end;

procedure TQueue.ThreadStarted;
begin
  InterlockedIncrement(FRunningThreads);
end;

消费者线程的代码简单明了.它标志着它的开始和结束,但这只是装饰性的,因为我希望能够显示正在运行的线程的数量,一旦创建所有线程就达到它的最大值,并且仅在第一个线程退出后才开始下降(是,当正在处理队列中的最后一批项目时).

{ TConsumer }

constructor TConsumer.Create(AQueue: TQueue);
// Context: calling thread.
begin
  inherited Create(False);
  FQueue := AQueue;
  // A consumer thread frees itself when the queue is emptied.
  FreeOnTerminate := True;
end;

procedure TConsumer.Execute;
// Context: This consumer thread
var
  Item: Integer;
  Value: String;
begin
  inherited;

  // Signal the queue (optional).
  FQueue.ThreadStarted;

  // Work until queue is empty (Pull returns false).
  while FQueue.Pull(Item, Value) do
  begin
    // Processing can take from .5 upto 1 second.
    Value := ReverseString(Value);
    Sleep(Random(500) + 1000);

    // Just save modified value back in queue.
    FQueue.Save(Item, Value);
  end;

  // Signal the queue (optional).
  FQueue.ThreadEnded;
end;

当然,如果要查看进度(或至少一点),则不需要阻止Run方法.或者,就像我一样,您可以在单独的线程中执行该阻塞方法:

  TQueueBackgroundThread = class(TThread)
  strict private
    FQueue: TQueue;
    FThreadCount: Integer;
  protected
    procedure Execute; override;
  public
    constructor Create(AQueue: TQueue; AThreadCount: Integer);
  end;

    { TQueueBackgroundThread }

constructor TQueueBackgroundThread.Create(AQueue: TQueue; AThreadCount: Integer);
begin
  inherited Create(False);
  FreeOnTerminate := True;
  FQueue := AQueue;
  FThreadCount := AThreadCount;
end;

procedure TQueueBackgroundThread.Execute;
// Context: This thread (TQueueBackgroundThread)
begin
  FQueue.Run(FThreadCount);
end;

现在,从GUI本身调用它.我创建了一个表单,它包含两个进度条,两个备忘录,一个计时器和一个按钮.Memo1充满了随机字符串.处理完成后,Memo2将接收处理过的字符串.计时器用于更新进度条,按钮是实际执行操作的唯一选择.

因此,表单只包含所有这些字段,以及对队列的引用.它还包含一个事件处理程序,以便在处理完成时得到通知:

type
  TForm1 = class(TForm)
    Button1: TButton;
    Memo1: TMemo;
    Memo2: TMemo;
    Timer1: TTimer;
    ProgressBar1: TProgressBar;
    ProgressBar2: TProgressBar;
    procedure Button1Click(Sender: TObject);
    procedure Timer1Timer(Sender: TObject);
  private
    Q: TQueue;
    procedure DoAllThreadsDone(Sender: TObject);
  end;

Button1单击事件,初始化GUI,创建包含100个项目的队列,并启动后台线程来处理队列.此后台线程接收OnTerminate事件处理程序(TThread的默认属性),以在处理完成时向GUI发送信号.

你可以在主线程中调用Q.Run,​​但它会阻止你的GUI.如果这是你想要的,那么你根本不需要这个线程!

procedure TForm1.Button1Click(Sender: TObject);
// Context: GUI thread
const
  ThreadCount = 10;
  StringCount = 100;
var
  i: Integer;
begin
  ProgressBar1.Max := ThreadCount;
  ProgressBar2.Max := StringCount;

  Memo1.Text := '';
  Memo2.Text := '';

  for i := 1 to StringCount do
    Memo1.Lines.Add(IntToHex(Random(MaxInt), 10));

  Q := TQueue.Create;
  Q.Items.Assign(Memo1.Lines);
  with TQueueBackgroundThread.Create(Q, ThreadCount) do
  begin
    OnTerminate := DoAllThreadsDone;
  end;
end;

处理线程完成时的事件处理程序.如果您希望处理阻止GUI,那么您不需要此事件处理程序,只需将此代码复制到Button1Click的末尾即可.

procedure TForm1.DoAllThreadsDone(Sender: TObject);
// Context: GUI thread
begin
  Memo2.Lines.Assign(Q.Items);
  FreeAndNil(Q);
  ProgressBar1.Position := 0;
  ProgressBar2.Position := 0;
end;

计时器仅用于更新进度条.它获取正在运行的线程数(只在处理几乎完成时才会下降),并且它获取"Item",这实际上是要处理的下一个项目.因此,当实际上最后10个项目仍在处理时,它可能已经完成.

procedure TForm1.Timer1Timer(Sender: TObject);
// Context: GUI thread
begin
  if Assigned(Q) then
  begin
    ProgressBar1.Position := Q.RunningThreads;
    ProgressBar2.Position := Q.Item;
    Caption := Format('%d, %d', [Q.RunningThreads, Q.Item]);
  end;
  Timer1.Interval := 20;
end;