multithreading - 线程同步队列的最佳方法

Question

我有一个队列，可以将不同的线程排入队列，因此我可以保证两件事:

1. 请求将被一一处理。
2. 请求按到达顺序处理

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

看起来像这样:

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

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

我需要这个，因为我有一个基于 Indy 的 Web 服务器，可以接受 http 请求。首先，我找到该请求的相应 session 。然后为该 session 执行请求(代码)。我可以收到同一 session 的多个请求(请阅读我可以在第一个请求仍在处理时收到新请求)，并且它们必须按照正确的到达顺序逐一执行。因此，我寻求一种可以在这种情况下使用的通用同步队列，以便可以对请求进行排队。我无法控制线程，每个请求可能在不同的线程中执行。

解决此类问题的最佳(常用)方法是什么？问题是入队和出队必须是原子操作，以便保留正确的顺序。我当前的实现有一个很大的瓶颈，但它有效。

编辑:以下是原子入队/出队操作的问题

你通常会做这样的事情:

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() 中，事件从列表顶部删除，并设置下一个事件(如果有)。

这样，最后一个请求线程将导致下一个请求线程解除阻塞，完全无需挂起或恢复线程。该解决方案也不需要任何额外的线程或窗口，每个请求线程只需要一个事件对象。