线程同步队列的最佳方法

Question

我有一个队列,我可以在其中排队不同的线程,所以我可以保证两件事:

1. 请求逐个处理.
2. 请求按到达顺序处理

第二点很重要.否则,一个简单的关键部分就足够了.我有不同的请求组,只有在一个组内才能满足这些要求.来自不同组的请求可以并发运行.

它看起来像这样:

FTaskQueue.Enqueu('MyGroup');
try
  Do Something (running in context of some thread)
finally
  FTaskQueue.Dequeu('MyGroup');
end;

编辑:我删除了实际的实现,因为它隐藏了我想要解决的问题

我需要这个,因为我有一个基于Indy的Web服务器接受http请求.首先,我找到了请求的相应会话.然后为该会话执行请求(代码).我可以获得同一会话的多个请求(读取我可以获得新请求,而第一个仍处理)并且它们必须以正确的到达顺序逐个执行.所以我寻找一个可以在这种情况下使用的通用同步队列,以便请求可以排队.我无法控制线程,每个请求可以在不同的线程中执行.

这种问题的最佳(使用)方法是什么？问题是Enqueue和Dequeue必须是原子操作,以便保存正确的顺序.我目前的实施有很大的瓶颈,但它确实有效.

编辑:Bellow是原子Enqueue/Dequeue操作的问题

你正常地做这样的事情:

procedure Enqueue;
begin
  EnterCriticalSection(FCritSec);
  try
    DoEnqueue;
  finally 
    LeaveCriticalSection(FCritSec);
  end;

  BlockTheCurrentThread; // here the thread blocks itself
end;

procedure Dequeue;
begin
  EnterCriticalSection(FCritSec);
  try
    DoDequeue;
    UnblockTheNextThread; // here the thread unblocks another thread
  finally 
    LeaveCriticalSection(FCritSec);
  end;
end;

现在问题在于这不是原子的.如果队列中已有一个线程而另一个线程调用Enqueue,则可能会发生第二个线程将离开临界区并尝试阻塞自身.现在线程调度程序将恢复第一个线程,它将尝试取消阻塞下一个(第二个)线程.但是第二个线程还没有被阻止,所以没有任何反应.现在第二个线程继续并阻塞自身,但这不正确,因为它不会被解除阻塞.如果阻塞在临界区内,那么临界区永远不会离开,我们就会陷入僵局.

Answer 1

另一种方法:

让每个请求线程都有一个最初未设置的手动重置事件.队列管理器是一个简单的对象,它维护这些事件的线程安全列表.的Enqueue()和Dequeue()方法均取请求线程作为参数的事件.

type
  TRequestManager = class(TObject)
  strict private
    fCritSect: TCriticalSection;
    fEvents: TList<TEvent>;
  public
    constructor Create;
    destructor Destroy; override;

    procedure Enqueue(ARequestEvent: TEvent);
    procedure Dequeue(ARequestEvent: TEvent);
  end;

{ TRequestManager }

constructor TRequestManager.Create;
begin
  inherited Create;
  fCritSect := TCriticalSection.Create;
  fEvents := TList<TEvent>.Create;
end;

destructor TRequestManager.Destroy;
begin
  Assert((fEvents = nil) or (fEvents.Count = 0));
  FreeAndNil(fEvents);
  FreeAndNil(fCritSect);
  inherited;
end;

procedure TRequestManager.Dequeue(ARequestEvent: TEvent);
begin
  fCritSect.Enter;
  try
    Assert(fEvents.Count > 0);
    Assert(fEvents[0] = ARequestEvent);
    fEvents.Delete(0);
    if fEvents.Count > 0 then
      fEvents[0].SetEvent;
  finally
    fCritSect.Release;
  end;
end;

procedure TRequestManager.Enqueue(ARequestEvent: TEvent);
begin
  fCritSect.Enter;
  try
    Assert(ARequestEvent <> nil);
    if fEvents.Count = 0 then
      ARequestEvent.SetEvent
    else
      ARequestEvent.ResetEvent;
    fEvents.Add(ARequestEvent);
  finally
    fCritSect.Release;
  end;
end;

每个请求线程调用Enqueue()队列管理器,然后等待自己的事件发出信号.然后它处理请求并调用Dequeue():

{ TRequestThread }

type
  TRequestThread = class(TThread)
  strict private
    fEvent: TEvent;
    fManager: TRequestManager;
  protected
    procedure Execute; override;
  public
    constructor Create(AManager: TRequestManager);
  end;

constructor TRequestThread.Create(AManager: TRequestManager);
begin
  Assert(AManager <> nil);
  inherited Create(TRUE);
  fEvent := TEvent.Create(nil, TRUE, FALSE, '');
  fManager := AManager;
  Resume;
end;

procedure TRequestThread.Execute;
begin
  fManager.Enqueue(fEvent);
  try
    fEvent.WaitFor(INFINITE);
    OutputDebugString('Processing request');
    Sleep(1000);
    OutputDebugString('Request processed');
  finally
    fManager.Dequeue(fEvent);
  end;
end;

{ TForm1 }

procedure TForm1.Button1Click(Sender: TObject);
var
  i: integer;
begin
  for i := 1 to 10 do
    TRequestThread.Create(fRequestManager);
end;

队列管理器锁定在两个事件列表Enqueue()和Dequeue().如果列表为空,Enqueue()则在参数中设置事件,否则重置事件.然后它将事件附加到列表中.因此,第一个线程可以继续请求,所有其他线程将阻止.在Dequeue()事件从列表顶部删除,并设置下一个事件(如果有).

这样,最后一个请求线程将导致下一个请求线程解除阻塞,完全没有挂起或恢复线程.此解决方案也不需要任何其他线程或窗口,每个请求线程只需要一个事件对象.