管理动态线程数

Question

首先，我对多线程还很熟悉，不懂太多术语。我需要确保我做得正确，因为这是一个敏感话题。

规格

我正在构建的是一个包含动态数量线程的组件。这些线程中的每一个都被重新用于执行多个请求。我可以在创建线程时和执行线程之前向其提供所有必要的详细信息，并提供事件处理程序。一旦执行，我就基本完成了一个请求，然后输入另一个请求。请求从另一个独立的后台线程输入到这些线程中，该线程不断处理请求队列。所以这个系统有两个列表：1）请求记录列表，2）线程指针列表。

我正在使用 TThread 类的后代（至少这是我熟悉的线程方法）。我通过同步创建线程时分配的事件触发器来从线程获取反馈。线程在后台加载和保存数据，完成后，它们会重置自己以准备处理下一个请求。

问题

现在，当决定如何处理更改允许线程数的事件时，麻烦就开始了（通过组件 ActiveThreads: TActiveThreadRange 的属性，其中 TActiveThreadRange代码> = 1..20)。因此，一次可以创建 1 到 20 个线程。但是，假设使用此组件的应用程序将此属性从 5 更改为 3。此时，已经创建了 5 个线程，如果该线程恰好很忙，我不想强​​制释放该线程。我需要等到它完成后再释放它。另一方面，如果属性从 3 更改为 5，那么我需要创建 2 个新线程。我需要知道在这种情况下“跟踪”这些线程的正确方法。

可能性

以下是我能想到的一些可能的方法来“跟踪”这些线程...

• 保留一个包含每个创建的线程的 TList - 易于管理
• 创建一个 TList 包含每个创建的线程的包装器或后代 - 更易于管理，但工作量更大
• 保留包含每个创建的线程的 array - 这会比 TList 更好吗？
• 创建一个包含每个创建的线程的数组包装器

但是回到我原来的问题 - 当 ActiveThreads 属性减少时如何处理现有的繁忙线程？创建它们没有问题，但释放它们却变得令人困惑。我通常制作可以自行释放的线程，但这是我第一次制作可以重复使用的线程。我只需要知道销毁这些线程而不中断其任务的正确方法。

更新

根据反馈，我已经获取并开始实现 OmniThreadLibrary（以及长期需要的 FastMM）。我还稍微改变了我的方法 - 一种可以创建这些线程进程而无需管理它们并且无需另一个线程来处理队列的方法...

• 1 个生成新进程的主方法
  • function NewProcess(const Request: TProcessRequest): TProcessInfo;
  • TProcessRequest 是一条记录，其中包含要执行的操作的规范（文件名、选项等）
  • TProcessInfo 是传回一些状态信息的记录。
• 创建新流程时，输入任务“完成”事件的事件处理程序。当组件收到此消息时，它将检查队列。
  • 如果命令已排队，它将比较活动进程限制与当前进程计数
  • >如果超出限制，则停止，下一个完成的进程将执行相同的检查
  • >如果在限制范围内，则启动另一个新流程（在确保完成前一个流程后）
  • 如果没有命令排队，则停止
• 每个进程在完成其任务后都会自行死亡（没有保持活动线程）
• 我不会必须担心另一个计时器或线程不断循环
  • 相反，每个进程都会销毁自己并在执行此操作之前检查新请求

另一个更新

我实际上已恢复使用TThread，因为 OTL 非常不舒服使用。我喜欢将事物包装并组织在自己的类中。

Answer 1

正如 @NGLN 等所解释的，您需要池化一些线程，并接受管理线程数的最简单方法是将实际线程数与所需数量分开。将线程添加到池中很容易 - 只需创建更多实例（将生产者-消费者任务输入队列作为参数传递，以便线程知道要等待什么）。如果所需的线程数少于当前现有的线程数，您可以排队足够的“毒丸”来杀死多余的线程。

不要保留任何线程指针列表 - 这是不必要的微观管理麻烦的负载（并且可能会出错）。您需要保留的只是池中所需线程数量的计数，以便您知道当“poolDepth”属性发生更改时要采取什么操作。

事件触发器最好加载到发布到池的作业中 - 它们全部来自某个“TpooledTask”类，该类将事件作为构造函数参数并将其存储在某个“FonComplete”TNotifyEvent 中。运行任务的线程可以在完成工作后调用 FonComplete（使用 TpooledTask 作为发送者参数） - 您不需要知道哪个线程运行该任务。

例子：

    unit ThreadPool;

    interface

    uses
      Windows, Messages, SysUtils, Variants, Classes, Graphics, Controls, Forms,
      Dialogs, StdCtrls, contnrs, syncobjs;


    type

    TpooledTask=class(TObject)
    private
      FonComplete:TNotifyEvent;
    protected
      Fparam:TObject;
      procedure execute; virtual; abstract;
    public
      constructor create(onComplete:TNotifyEvent;param:TObject);
    end;

    TThreadPool=class(TObjectQueue)
    private
      access:TcriticalSection;
      taskCounter:THandle;
      threadCount:integer;
    public
      constructor create(initThreads:integer);
      procedure addTask(aTask:TpooledTask);
    end;

    TpoolThread=class(Tthread)
    private
      FmyPool:TThreadPool;
    protected
      procedure Execute; override;
    public
      constructor create(pool:TThreadPool);
    end;

    implementation

    { TpooledTask }

    constructor TpooledTask.create(onComplete: TNotifyEvent; param: TObject);
    begin
      FonComplete:=onComplete;
      Fparam:=param;
    end;

    { TThreadPool }

    procedure TThreadPool.addTask(aTask: TpooledTask);
    begin
      access.acquire;
      try
        push(aTask);
      finally
        access.release;
      end;
      releaseSemaphore(taskCounter,1,nil); // release one unit to semaphore
    end;

    constructor TThreadPool.create(initThreads: integer);
    begin
      inherited create;
      access:=TcriticalSection.create;
      taskCounter:=createSemaphore(nil,0,maxInt,'');
      while(threadCount<initThreads) do
      begin
        TpoolThread.create(self);
        inc(threadCount);
      end;
    end;

    { TpoolThread }

    constructor TpoolThread.create(pool: TThreadPool);
    begin
      inherited create(true);
      FmyPool:=pool;
      FreeOnTerminate:=true;
      resume;
    end;

procedure TpoolThread.execute;
var thisTask:TpooledTask;
begin
  while (WAIT_OBJECT_0=waitForSingleObject(FmyPool.taskCounter,INFINITE)) do
  begin
    FmyPool.access.acquire;
    try
      thisTask:=TpooledTask(FmyPool.pop);
    finally
      FmyPool.access.release;
    end;
    thisTask.execute;
    if assigned(thisTask.FonComplete) then thisTask.FonComplete(thisTask);
  end;
end;

end.

Answer 2

您可以在请求队列中实现 FreeNotify 消息，当工作线程收到此消息时释放自己。在您的示例中，当您将线程数从 5 减少到 3 时，只需将 2 个 FreeNotify 消息放入队列中，然后 2 个工作线程将空闲。

Answer 3

关于活动线程减少的问题：抱歉，但您只需自己决定。要么立即释放不需要的线程（尽早终止它们），要么让它们运行直到完成（所有工作完成后终止它们）。这是你的选择。当然，您必须将所需数量的变量与实际线程数的变量分开。更新线程变量的实际数量（可能只是 List.Count）的问题对于两者来说完全相同，因为任一解决方案都需要一些时间。

关于多线程的管理：您可以研究这个答案，它将线程存储在 TList 中。不过，它需要根据您的具体愿望清单进行一些调整，如果需要帮助，请大声喊叫。此外，当然还有更多可能的实现可以从默认 TThread 的使用中派生出来。请注意，存在其他（多）线程库，但我从未需要使用它们。