如何使用 OmniThreadLibrary 监控特定任务

Question

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

我有一个方法 ValidateInvoiceFile。发票文件只是一个独立的发票对象列表。

ValidateInvoiceFile 在我的服务中被调用。用户 1 验证文件 1，用户 2 验证文件 2 - 这将发生在 2 个完全独立的调用中

例如

用户 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;