multithreading - 管理动态线程数

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 使用起来非常不舒服。我喜欢在自己的类(class)中进行包装和组织。

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.