如何在Delphi中用队列实现多线程，限制并发线程数为2？

需求 #

• 同一时间仅允许2个线程运行
• 未执行的任务进入队列等待，线程资源释放后自动执行

已尝试的方案及问题 #

1. TParallel.For #

TParallel.For默认根据系统CPU核心数调度线程，无法直接限制并发数到2，因此无法满足需求。示例代码：

type
  TMyRecord = record
    msg: String;
    sleep: Integer;
  end;

var
  Form2: TForm2;

implementation

{$R *.dfm}

procedure TForm2.Button1Click(Sender: TObject);
var
  record1, record2, record3, record4, record5: TMyRecord;
  MyList: TList<TMyRecord>;
begin
  // Initialize records
  record1.msg := 'Item1';
  record1.sleep := 10;

  record2.msg := 'Item2';
  record2.sleep := 10;

  record3.msg := 'Item3';
  record3.sleep := 3000;

  record4.msg := 'Item4';
  record4.sleep := 3000;

  record5.msg := 'Item5';
  record5.sleep := 100;

  MyList := TList<TMyRecord>.Create;
  try
    // Add records to the list
    MyList.Add(record1);
    MyList.Add(record2);
    MyList.Add(record3);
    MyList.Add(record4);
    MyList.Add(record5);

    // Use TParallel.For
    TParallel.For(0, MyList.Count - 1,
      procedure (i: Integer)
      var
        ListItem: TMyRecord;
      begin
        ListItem := MyList[i];

        TThread.Synchronize(nil,
          procedure
          begin
            Memo1.Lines.Add(ListItem.msg);
          end);

        Sleep(ListItem.sleep);
      end);

  finally
    MyList.Free;
  end;
end;

2. TThreadPool #

使用TThreadPool时出现两个问题：

• 仅重复打印Item5五次
• Sleep等待时间未生效

问题原因：

1. 循环中var ListItem := MyList[i]的变量被所有匿名函数共享，循环结束后变量指向最后一个元素，导致所有任务处理Item5；
2. Sleep被放在TThread.Queue的匿名函数中，实际在主线程执行，不会阻塞工作线程，还会导致主线程卡顿。

示例代码：

procedure TForm2.Button1Click(Sender: TObject);
var
  record1, record2, record3, record4, record5: TMyRecord;
  MyList: TList<TMyRecord>;
  ThreadPool: TThreadPool;
begin
  // Initialize records
  record1.msg := 'Item1';
  record1.sleep := 10;

  record2.msg := 'Item2';
  record2.sleep := 10;

  record3.msg := 'Item3';
  record3.sleep := 3000;

  record4.msg := 'Item4';
  record4.sleep := 3000;

  record5.msg := 'Item5';
  record5.sleep := 100;

  MyList := TList<TMyRecord>.Create;
  try
    // Add records to the list
    MyList.Add(record1);
    MyList.Add(record2);
    MyList.Add(record3);
    MyList.Add(record4);
    MyList.Add(record5);

    ThreadPool := TThreadPool.Create;
    ThreadPool.SetMaxWorkerThreads(2);

    // Iterate through the list and queue each work item with a copy of the record
    for var i := 0 to MyList.Count - 1 do
    begin
      var ListItem := MyList[i];

      ThreadPool.QueueWorkItem(
        procedure
        var
          LocalItem: TMyRecord;
        begin
          LocalItem := ListItem;

          TThread.Queue(nil,
            procedure
            begin
              Memo1.Lines.Add(LocalItem.msg);
              Sleep(LocalItem.sleep);  // Simulate the delay
            end);
        end);
    end;
  finally
    MyList.Free;
  end;
end;

解决方案 #

方案1：修正TThreadPool用法 #

解决变量捕获问题，将耗时操作放在工作线程，仅UI更新提交到主线程：

procedure TForm2.Button1Click(Sender: TObject);
var
  record1, record2, record3, record4, record5: TMyRecord;
  MyList: TList<TMyRecord>;
  ThreadPool: TThreadPool;
begin
  // Initialize records
  record1.msg := 'Item1';
  record1.sleep := 10;

  record2.msg := 'Item2';
  record2.sleep := 10;

  record3.msg := 'Item3';
  record3.sleep := 3000;

  record4.msg := 'Item4';
  record4.sleep := 3000;

  record5.msg := 'Item5';
  record5.sleep := 100;

  MyList := TList<TMyRecord>.Create;
  try
    MyList.Add(record1);
    MyList.Add(record2);
    MyList.Add(record3);
    MyList.Add(record4);
    MyList.Add(record5);

    ThreadPool := TThreadPool.Create;
    try
      ThreadPool.SetMaxWorkerThreads(2);
      ThreadPool.SetMinWorkerThreads(2);

      for var i := 0 to MyList.Count - 1 do
      begin
        const CurrentItem = MyList[i];
        ThreadPool.QueueWorkItem(
          procedure
          begin
            Sleep(CurrentItem.sleep);
            TThread.Queue(nil,
              procedure
              begin
                Memo1.Lines.Add(Format('完成：%s', [CurrentItem.msg]));
              end);
          end);
      end;
    finally
      // 若需等待所有任务完成，可调用ThreadPool.WaitFor后再Free
      // ThreadPool.WaitFor;
      // ThreadPool.Free;
    end;
  finally
    MyList.Free;
  end;
end;

方案2：信号量（TSemaphore）配合TTask #

通过信号量严格控制并发数，最多允许2个线程同时执行：

uses
  System.SyncObjs;

procedure TForm2.Button1Click(Sender: TObject);
var
  record1, record2, record3, record4, record5: TMyRecord;
  MyList: TList<TMyRecord>;
  Semaphore: TSemaphore;
begin
  // Initialize records
  record1.msg := 'Item1';
  record1.sleep := 10;

  record2.msg := 'Item2';
  record2.sleep := 10;

  record3.msg := 'Item3';
  record3.sleep := 3000;

  record4.msg := 'Item4';
  record4.sleep := 3000;

  record5.msg := 'Item5';
  record5.sleep := 100;

  MyList := TList<TMyRecord>.Create;
  try
    MyList.Add(record1);
    MyList.Add(record2);
    MyList.Add(record3);
    MyList.Add(record4);
    MyList.Add(record5);

    Semaphore := TSemaphore.Create(nil, 2, 2, '');
    try
      for var Item in MyList do
      begin
        const CurrentItem = Item;
        TTask.Run(
          procedure
          begin
            try
              Semaphore.Acquire;
              Sleep(CurrentItem.sleep);
              TThread.Queue(nil,
                procedure
                begin
                  Memo1.Lines.Add(Format('完成：%s', [CurrentItem.msg]));
                end);
            finally
              Semaphore.Release;
            end;
          end);
      end;
    finally
      Semaphore.Free;
    end;
  finally
    MyList.Free;
  end;
end;

说明 #

• 方案1通过TThreadPool的线程数限制实现需求，适合依赖线程池管理的场景；
• 方案2使用信号量是通用的并发控制方式，适用于任何需要限制并发数的多线程场景；
• 所有VCL控件操作必须通过TThread.Queue或TThread.Synchronize提交到主线程，避免跨线程访问异常。

内容的提问来源于stack exchange，提问作者I'mSRJ