这是我如何编写消费者线程。拼图的第一部分是一个阻塞队列。我看起来是这样的：
unit BlockingQueue;
interface
uses
Windows, SyncObjs, Generics.Collections;
type
TBlockingQueue<T> = class
//see Duffy, Concurrent Programming on Windows, pp248
private
FCapacity: Integer;
FQueue: TQueue<T>;
FLock: TCriticalSection;
FNotEmpty: TEvent;
function DoEnqueue(const Value: T; IgnoreCapacity: Boolean): Boolean;
public
constructor Create(Capacity: Integer=-1);//default to unbounded
destructor Destroy; override;
function Enqueue(const Value: T): Boolean;
procedure ForceEnqueue(const Value: T);
function Dequeue: T;
end;
implementation
{ TBlockingQueue<T> }
constructor TBlockingQueue<T>.Create(Capacity: Integer);
begin
inherited Create;
FCapacity := Capacity;
FQueue := TQueue<T>.Create;
FLock := TCriticalSection.Create;
FNotEmpty := TEvent.Create(nil, True, False, '');
end;
destructor TBlockingQueue<T>.Destroy;
begin
FNotEmpty.Free;
FLock.Free;
FQueue.Free;
inherited;
end;
function TBlockingQueue<T>.DoEnqueue(const Value: T; IgnoreCapacity: Boolean): Boolean;
var
WasEmpty: Boolean;
begin
FLock.Acquire;
Try
Result := IgnoreCapacity or (FCapacity=-1) or (FQueue.Count<FCapacity);
if Result then begin
WasEmpty := FQueue.Count=0;
FQueue.Enqueue(Value);
if WasEmpty then begin
FNotEmpty.SetEvent;
end;
end;
Finally
FLock.Release;
End;
end;
function TBlockingQueue<T>.Enqueue(const Value: T): Boolean;
begin
Result := DoEnqueue(Value, False);
end;
procedure TBlockingQueue<T>.ForceEnqueue(const Value: T);
begin
DoEnqueue(Value, True);
end;
function TBlockingQueue<T>.Dequeue: T;
begin
FLock.Acquire;
Try
while FQueue.Count=0 do begin
FLock.Release;
Try
FNotEmpty.WaitFor;
Finally
FLock.Acquire;
End;
end;
Result := FQueue.Dequeue;
if FQueue.Count=0 then begin
FNotEmpty.ResetEvent;
end;
Finally
FLock.Release;
End;
end;
end.
它完全是线程安全的。任何线程都可以入队。任何线程都可以出队。如果队列为空，则出队功能将被阻塞。队列可以以有界模式或无界模式运行。
接下来，我们需要一个与这样的队列一起工作的线程。线程简单地将作业从队列中取出，直到它被告知终止。我的消费线程如下所示：
unit ConsumerThread;
interface
uses
SysUtils, Classes, BlockingQueue;
type
TConsumerThread = class(TThread)
private
FQueue: TBlockingQueue<TProc>;
FQueueFinished: Boolean;
procedure SetQueueFinished;
protected
procedure TerminatedSet; override;
procedure Execute; override;
public
constructor Create(Queue: TBlockingQueue<TProc>);
end;
implementation
{ TConsumerThread }
constructor TConsumerThread.Create(Queue: TBlockingQueue<TProc>);
begin
inherited Create(False);
FQueue := Queue;
end;
procedure TConsumerThread.SetQueueFinished;
begin
FQueueFinished := True;
end;
procedure TConsumerThread.TerminatedSet;
begin
inherited;
//ensure that, if the queue is empty, we wake up the thread so that it can quit
FQueue.ForceEnqueue(SetQueueFinished);
end;
procedure TConsumerThread.Execute;
var
Proc: TProc;
begin
while not FQueueFinished do begin
Proc := FQueue.Dequeue();
Proc();
Proc := nil;//clear Proc immediately, rather than waiting for Dequeue to return since it blocks
end;
end;
end.
这具有您正在寻找的属性。也就是说，当线程被销毁时，它会在完成析构函数之前处理所有挂起的任务。
要看到它的行动，这里是一个简短的演示程序：
unit Main;
interface
uses
Windows, SysUtils, Classes, Controls, Forms, StdCtrls,
BlockingQueue, ConsumerThread;
type
TMainForm = class(TForm)
Memo1: TMemo;
TaskCount: TEdit;
Start: TButton;
Stop: TButton;
procedure StartClick(Sender: TObject);
procedure StopClick(Sender: TObject);
private
FQueue: TBlockingQueue<TProc>;
FThread: TConsumerThread;
procedure Proc;
procedure Output(const Msg: string);
end;
implementation
{$R *.dfm}
procedure TMainForm.Output(const Msg: string);
begin
TThread.Synchronize(FThread,
procedure
begin
Memo1.Lines.Add(Msg);
end
);
end;
procedure TMainForm.Proc;
begin
Output(Format('Consumer thread ID: %d', [GetCurrentThreadId]));
Sleep(1000);
end;
procedure TMainForm.StartClick(Sender: TObject);
var
i: Integer;
begin
Memo1.Clear;
Output(Format('Main thread ID: %d', [GetCurrentThreadId]));
FQueue := TBlockingQueue<TProc>.Create;
FThread := TConsumerThread.Create(FQueue);
for i := 1 to StrToInt(TaskCount.Text) do
FQueue.Enqueue(Proc);
end;
procedure TMainForm.StopClick(Sender: TObject);
begin
Output('Stop clicked, calling thread destructor');
FreeAndNil(FThread);
Output('Thread destroyed');
FreeAndNil(FQueue);
end;
end.
object MainForm: TMainForm
Caption = 'MainForm'
ClientHeight = 560
ClientWidth = 904
object Memo1: TMemo
Left = 0
Top = 96
Width = 904
Height = 464
Align = alBottom
end
object TaskCount: TEdit
Left = 8
Top = 8
Width = 121
Height = 21
Text = '10'
end
object Start: TButton
Left = 8
Top = 48
Width = 89
Height = 23
Caption = 'Start'
OnClick = StartClick
end
object Stop: TButton
Left = 120
Top = 48
Width = 75
Height = 23
Caption = 'Stop'
OnClick = StopClick
end
end
权，但你可以在上面我的代码看，我在一个循环'，而不是终止...' – 2013-02-22 15:44:24
所以，Terminated将被设置为True，然后Execute将返回。 – 2013-02-22 15:45:18
但是仍然有剩余的项目需要保存，循环需要继续，直到队列为空。 – 2013-02-22 15:46:01