如何在OmniThreadLibrary的Parallel.ForEach中接收单个任务完成的通知

看起来你现在的问题是把消息发送逻辑放在了所有任务执行完毕的环节，所以才只会收到一次最终通知。其实要实现单个任务完成就触发通知，很简单——只需要把消息发送的代码放到每个任务的执行过程内部即可。

你的Parallel.ForEach的Execute方法里，每个任务都会单独执行那个匿名函数，我们可以在每个任务完成自身工作（创建好TTransaction）之后，立刻通过当前任务的Comm通道发送消息给主线程，这样每个任务结束时都会触发一次你的MSGNewData处理函数。

修改后的代码示例如下：

procedure TfrmForEachOutput.ProcessTransactions(input: TList<TSSRINTBWTH>; output: TList<TTransaction>);
var
  outQueue   : IOmniBlockingCollection;
  transaction: TOmniValue;
begin
  outQueue := TOmniBlockingCollection.Create;
  Parallel.ForEach(0, input.Count - 1)
    .TaskConfig(Parallel.TaskConfig.OnMessage(Self))
    .NoWait
    .Into(outQueue)
    .Execute(
      procedure(const task: IOmniTask; const value: integer; var result: TOmniValue)
      begin
        result := TTransaction.Create(input[value], task);
        // 每个任务完成后立即发送通知
        task.Comm.Send(MSG_NEWDATA, Format('索引 %d 的任务已完成', [value]));
      end
    );

  for transaction in outQueue do
  begin
    output.Add(transaction.AsObject as TTransaction);
  end;
end;

你还可以调整MSGNewData处理函数，方便直观看到具体是哪个任务完成了：

procedure TfrmForEachOutput.MSGNewData(var msg: TOmniMessage);
begin
  OutputDebugString(PChar(Format('收到任务完成通知：%s', [msg.MsgData.AsString])));
end;

这里的核心逻辑是：每个任务实例（task: IOmniTask）都自带线程安全的通信通道Comm，通过它发送的消息会被你在TaskConfig里指定的接收者（也就是你的窗体Self）处理，而且是每个任务完成就立刻发送，不会等到所有任务都结束。

如果需要在消息里传递更复杂的数据，也可以用TOmniValue封装，它能安全地跨线程传递各种类型的数据，避免线程安全问题。

备注：内容来源于stack exchange，提问作者chang long