主页 / coding / 问题 / 79318289
Accepted
I'mSRJ
Asked: 2024-12-31 02:10:21 +0800 CST

在 Delphi 中使用队列实现多线程以限制同时运行的 n 个线程？

  • 772

我已经使用 Delphi 有一段时间了，现在正尝试在我的代码中实现多线程，要求如下：

  1. 每次只能运行 2 个线程。

  2. 其他任务应在队列中等待，并在线程可用时执行。

我尝试使用TParallel.For来实现这一点，但无法实现所需的行为。这是我的示例代码：

    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;

我也尝试使用TTask.Run和ThreadPool，但无法实现所需的行为。更新 - 这是我尝试使用 Threadpool 执行的操作

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. 它仅打印“item5”5次。
  2. 等待时间（睡眠）不起作用。

我该如何修改此代码以将同时运行的线程数限制为 2？并在线程可用时将其他任务排队执行？

multithreading

2 个回答

  1. Best Answer

    它仅打印“item5”5次。

    这意味着您排队的工作项正在捕获相同的内容TMyRecord，并且所有打印都是在该变量被分配列表中的最后一项之后发生的。

    即使您的循环正在使用本地内联变量，但考虑到匿名过程变量捕获的工作方式，它似乎忽略了内联。

    让匿名过程捕获循环内的变量的常见解决方案是将匿名过程移动到具有可捕获的输入参数的单独函数。

    等待时间（睡眠）不起作用。

    这可能是因为您是在主 UI 线程的上下文中执行睡眠，而不是在工作线程中，所以它们全速运行。

    尝试一些更像这样的事情：

    function MakeWorkItem(Item: TMyRecord): TProc;
begin
  Result := procedure
    begin
      TThread.Queue(nil,
        procedure
        begin
          Form2.Memo1.Lines.Add(Item.msg);
        end
      );
      Sleep(Item.sleep);  // Simulate the delay 
    end;
end; 

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
      ThreadPool.QueueWorkItem(MakeWorkItem(MyList[i]));
    end;
  finally
    MyList.Free;
  end;
end;

    或者，使用TParallel.For，只需将您想要的内容TThreadPool作为额外参数传递给它，例如：

    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);

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

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

        Sleep(ListItem.sleep);
      end,
      ThreadPool
    );
  finally
    MyList.Free;
  end;
end;
    • 3

  2. 您可以相当容易地使用专为此任务设计的TThreadedQueue 来完成此任务。

    我整理了一个小演示： https: //github.com/ideasawakened/DelphiKB/tree/master/Delphi%20Demos/TThreadedQueue/TwoClientThreads_MainThreadProducer

    • 创建一个TThreadedQueue
    • 衍生 2 个客户端工作线程
    • 单击按钮生成 TMyRecords 并使用 PushItem 将它们添加到队列中，然后 2 个工作线程将按顺序完成它们。

    示例代码。（+ 向表单添加 TButton/TMemo...）

    unit Unit1;

interface

uses
  Vcl.Controls,
  Vcl.Forms,
  Vcl.StdCtrls,
  System.Classes,
  System.Generics.Collections;

type

  TMyRecord = record
    id:Integer;
    msg:string;
    sleep:Integer;
  end;

  TMyRecConsumerThread = class(TThread)
  private
    fQueue:TThreadedQueue<TMyRecord>;
    fThreadNumber:Integer;
  public
    constructor Create(const Queue:TThreadedQueue<TMyRecord>; const ThreadNumber:Integer);

    procedure Execute(); override;
  end;


  TMainForm = class(TForm)
    Memo1:TMemo;
    butProduce:TButton;
    procedure butProduceClick(Sender:TObject);
    procedure FormCreate(Sender:TObject);
    procedure FormDestroy(Sender:TObject);
  private const
    ConsumerThreadCount = 2;

    QueueDepth = 100;
    ProduceCount = 5;
    PushTimeout = 500;
    PopTimeout = 3000;
  private
    fQueue:TThreadedQueue<TMyRecord>;
    fRecId:Integer;
    fConsumers:TArray<TMyRecConsumerThread>;
    function CreateWorkItem:TMyRecord;
  end;

var
  MainForm:TMainForm;

implementation

uses
  System.SysUtils,
  System.SyncObjs;

{$R *.dfm}


procedure TMainForm.FormCreate(Sender:TObject);
begin
  ReportMemoryLeaksOnShutdown := True;
  fRecId := 1;
  butProduce.Caption := Format('Produce %d records', [ProduceCount]);

  fQueue := TThreadedQueue<TMyRecord>.Create(QueueDepth, PushTimeout, PopTimeout);

  SetLength(fConsumers, ConsumerThreadCount);
  for var i := 1 to ConsumerThreadCount do
  begin
    fConsumers[i-1] := TMyRecConsumerThread.Create(fQueue, i);
  end;
end;


procedure TMainForm.FormDestroy(Sender:TObject);
begin
  for var ConsumerThread in fConsumers do
  begin
    ConsumerThread.Free;
  end;
  fQueue.DoShutDown;
  fQueue.Free;
end;


function TMainForm.CreateWorkItem:TMyRecord;
begin
  Result := Default(TMyRecord);
  Result.id := fRecId;
  Result.sleep := Random(PopTimeout div 2);
  Result.msg := Format('Task %d sleeping %dms', [fRecId, Result.sleep]);

  Memo1.Lines.Add(Format('Produced Task %d to sleep %dms', [fRecId, Result.sleep]));
  Inc(fRecId);
end;


//click the button one or more times to queue up some work items
procedure TMainForm.butProduceClick(Sender:TObject);
var
  wr:TWaitResult;
begin
  for var i := 1 to ProduceCount do
  begin
    wr := fQueue.PushItem(CreateWorkItem);
    Assert(wr = TWaitResult.wrSignaled);  // IRL deal with <> wrSignaled (e.g., timeout if queue is full)
  end;
end;


constructor TMyRecConsumerThread.Create(const Queue:TThreadedQueue<TMyRecord>; const ThreadNumber:Integer);
begin
  fQueue := Queue;
  fThreadNumber := ThreadNumber;
  inherited Create({CreateSuspended=}False);
end;


procedure TMyRecConsumerThread.Execute;
var
  wr:TWaitResult;
  r:TMyRecord;
begin
  NameThreadForDebugging(Format('Consumer%d', [fThreadNumber]));

  while (not Terminated) and (not fQueue.ShutDown) do
  begin
    wr := fQueue.PopItem(r);
    if (wr = wrSignaled) then
    begin
      if r.id > 0 then //Note: if 0, then PopItem was signalled as Queue was told to shut down (rec will be "Default(TMyRecord)")
      begin
        TThread.Synchronize(nil,
          procedure
          begin
            MainForm.Memo1.Lines.Add(Format('Consumed %s', [r.msg]));
          end);
        sleep(r.sleep); //would be better to have an abortable sleep using an TEvent
      end;
    end;
  end;
end;


end.
    • 2

相关问题