如何使用OmniThreadLibrary监视特定任务

Question

我想使用OmniThreadLibrary的默认线程池，即不创建我自己的本地线程池。

发票文件只是一个自包含的发票对象列表。

在我的服务中调用ValidateInvoiceFile。

例如

用户1 -文件1-发票1、发票2、发票3

用户2-文件2-发票34231，发票31235

每张发票都将在其自己的任务中进行验证。所以在这种情况下，我将有5个任务。

我需要我的ValidateInvoiceFiles过程等待，直到创建的所有任务都完成。不是所有的任务。

文件1的ValidateInvoices调用对文件2的任务不感兴趣，因此我不希望它等待它们

我该如何使用Delphi的OmniThreadLibrary来做这件事呢

我在想OnThreadTerminated事件，但我不确定如何让它只关心在特定调用期间添加的任务。

procedure ValidateInvoiceFile(const objInvoices: TInvoices)
begin
  try
    objThreadPool := CreateThreadPool('Connection pool');
    objThreadPool.MaxExecuting := 2 * System.CPUCount;

    nStart := 0;
    nEnd := 49;

    while nStart <= objInvoices.Count - 1 do
    begin
      for InvoiceIndex := nStart to nEnd do
      begin
        if InvoiceIndex > objInvoices.Count - 1 then
          Break;

        objInvoice := objInvoices[InvoiceIndex];

        objValidationData := TValidationData.Create(FConnection, FAllInvoices, FAllInvoices[InvoiceIndex]);
        //I fill objValidationData with information to be given to the task

        CreateTask(
          procedure(const task: IOmniTask)
          var
            objLocalInvoice: TInvoice;
            ValidatorIndex: Integer;
            objValidator:   TInvValidator;
            objUtilities:   TUtilities;
            objValidationData: TValidationData;
            nLocalInvoiceIndex: Integer;
            nValidatorIndex: Integer;
          begin
            nLocalInvoiceIndex := Task.Param['InvoiceIndex'].AsInteger;
            objValidationData := TValidationData(Task.Param['objValidationData'].AsObject);
            try
               //Do my work here
            finally
              FreeAndNil(objValidationData);
            end;
          end
          )
          .SetParameter('InvoiceIndex', InvoiceIndex)
          .SetParameter('objValidationData', TObject(objValidationData))
          .Unobserved
          .Schedule(objThreadPool);
      end;

      /**** This is the part I dont know how to change*****/
      while objThreadPool.CountExecuting + objThreadPool.CountQueued > 0 do
        Self := Self;

        Inc(nStart, 50);
        Inc(nEnd, 50);
     end;
  finally
    objThreadPool := nil;
  end;
end;

Answer 1

类似这样的东西应该可以工作(如果我正确理解了这个问题)：

uses   
  OtlCommon,   
  OtlParallel;

GlobalParallelPool.MaxExecuting := Environment.System.Affinity.Count * 2;

procedure ValidateInvoiceFile(const objInvoices: TInvoices);
begin   
  Parallel.ForEach(0, objInvoices.Count - 1)
    .Execute(
      procedure(const invoiceIndex: integer)
      begin
        //validate objInvoices[invoiceIndex]
      end); 
end;