如何在 Delphi 中使用流水线模式

Question

我正在尝试在我的测试项目 (How to make a Mutlithreded idhttp calls to do work on a StringList) 中实现流水线模式，但在将 TThread 代码适应流水线模式代码时遇到了困难。关于如何使用它的资源并不多。

我在下面尽力了，请不要投反对票，我知道我的代码很乱，但如果需要我会编辑我的问题。

type
  TForm2 = class(TForm)
    ...
  private
    procedure Retriever(const input: TOmniValue; var output: TOmniValue);
    procedure Inserter(const input, output: IOmniBlockingCollection);
    function HttpGet(url: string; var page: string): boolean;
  end;

procedure TForm2.startButton1Click(Sender: TObject);
var
  pipeline: IOmniPipeline;
  i       : Integer;
  v       : TOmniValue;
  s       : string;
  urlList : TStringList;
begin
  pipeline := Parallel.Pipeline;
  pipeline.Stage(Retriever);
  pipeline.Stage(Inserter).NumTasks(10);
  pipeline.Run;
  for s in urlList do
    pipeline.Input.Add(s);
  pipeline.Input.CompleteAdding;
  // wait for pipeline to complete
  pipeline.WaitFor(INFINITE);
end;

function TForm2.HttpGet(url: string; var page: string): boolean;
var
  lHTTP: TIdHTTP;
  i : integer;
  X : Tstrings;
  S,M,fPath : String;
begin
  lHTTP := TIdHTTP.Create(nil);
  X := TStringList.Create;
  try
    X.Text := lHTTP.Get('https://instagram.com/'+fPath);
    S:= ExtractDelimitedString(X.Text);
    X.Clear;
    Memo2.Lines.Add(fPath+ ' :     '+ M ); //how to pass the result to Inserter
  finally
    lHttp.Free;
  end;
end;

procedure TForm2.Inserter(const input, output: IOmniBlockingCollection);
var
  result   : TOmniValue;
  lpage     : string;
begin
  for result in input do begin
    Memo2.Lines.Add(lpage);
    FreeAndNil(lpage);
  end;
  // correect?
end;

procedure TForm2.Retriever(const input: TOmniValue; var output: TOmniValue);
var
  pageContents: string;
begin
  if HttpGet(input.AsString, pageContents) then
    output := //???
end;

Answer 1

首先 - 描述您的具体问题。没有人可以站在你背后看着你的电脑，看看你在做什么。 http://www.catb.org/esr/faqs/smart-questions.html#beprecise

您确实暗示您的程序行为不端。但是你没有描述如何以及为什么。而我们并不知道。

作为一般评论，您有点过度使用管道。

1. 您传递给 OTL 的所有工作程序 - 在您的情况下，这些是 Inserter 和 Retriever 在随机线程中工作。这意味着没有synchronizing，它们都不应该接触GUI - VCL 不是多线程的。 正如我在链接问题中向您解释的那样，使用TThread.Synchronize 也是一个糟糕的选择。它使程序变慢并且使表单不可读。要更新您的表单，请使用固定帧速率的轮询。不要从 OTL 工作人员内部更新您的表单。

换句话说，Inserter 不是您所需要的。您需要从管道中获取的只是它的输入集合、下载程序和输出集合。是的，对于复杂的管道来说，这是非常简单的任务，这就是为什么我在它之前提到了另外两个更简单的模式。

您的表单上需要TTimer，它将以每秒 2-3 次固定帧速率轮询输出集合，并检查该集合是否尚未最终确定（如果是 - 管道已停止）并且应该更新来自主线程的 GUI。

1. 您不应等待管道在您的主 VCL 线程内完成。相反，您应该分离管道并让它完全在后台运行。将对创建的管道的引用保存到 Form 的成员变量中，以便您可以从 TTimer 事件访问其输出集合，并且还可以在其进程运行后释放管道。

您应该保持该变量链接到管道对象，直到下载结束，然后设置为nil（释放对象），而不是之前。 你知道 Delphi 中的接口和引用计数，对吗？

对于其他 OTL 模式，例如 parallel-FOR，请阅读有关其 .NoWait() 调用的 OTL 文档。

1. 您应该将此表单设置为双模式，以便在下载运行时和未运行时启用不同的控件集。我通常使用特殊的布尔属性来做到这一点，就像我在您链接的主题中向您展示的那样。 您的用户不应该在管道正在进行时更改列表和设置（除非您要实现实时任务更改，但您还没有）。当从工作模式切换到空闲模式时，此模式切换器也是释放已完成管道对象的好地方。

1234563 >Unpacker 阶段从输入集合中获取字符串数组（实际上只有一个）并枚举它并将字符串放入阶段输出集合中。 然而，这几乎没有什么实用价值，它甚至会减慢你的程序速度，因为Memo1.Lines.ToArray() 函数仍然可以在主 VCL 线程中工作。但只是为了试验管道，这可能很有趣。

所以草稿变成了这样，

 TfrmMain = class(TForm)
  private
    var pipeline: IOmniPipeline;

    property inProcess: Boolean read ... write SetInProcess;
...
  end.

procedure Retriever(const input: TOmniValue; var output: TOmniValue);
var
  pageContents, URL: string;
  lHTTP: TIdHTTP;
begin
  URL := input.AsString;

  lHTTP := TIdHTTP.Create(nil);
  try
    lHTTP.ReadTimeout := 30000;
    lHTTP.HandleRedirects := True;

    pageContents := ExtractDelimitedString( lHTTP.Get('https://instagram.com/' + URL) );

    if pageContents > '' then
       Output := pageContents;
  finally
    lHTTP.Destroy;
  end;
end;

procedure TfrmMain.FormCloseQuery(Sender: TObject; var CanClose: Boolean);
begin
  if InProgress then begin
     CanClose := False;
     ShowMessage( 'You cannot close this window now.'^M^J+
                  'Wait for downloads to complete first.' ); 
  end;
end;

procedure TfrmMain.SetInProcess(const Value: Boolean);
begin
  if Value = InProcess then exit; // form already is in this mode

  FInProcess := Value;

  memo1.ReadOnly := Value;
  StartButton.Enabled := not Value;
  if Value then 
     Memo2.Lines.Clear;

  Timer1.Delay := 500; // twice per second
  Timer1.Enabled := Value;

  If not Value then  // for future optimisation - make immediate mode change 
     FlushData;      // when last worker thread quits, no waiting for timer event

  If not Value then
     pipeline := nil; // free the pipeline object

  If not Value then
     ShowMessage('Work complete');
end;

procedure TfrmMain.Timer1Timer(const Sender: TObject);
begin
  If not InProcess then exit;

  FlushData;

  if Pipeline.Output.IsFinalized then
     InProcess := False;
end;

procedure TForm2.startButton1Click(Sender: TObject);
var
  s       : string;
  urlList : TStringList;
begin
  urlList := Memo1.Lines;

  pipeline := Parallel.Pipeline;

  pipeline.Stage(Retriever).NumTasks(10).Run;

  InProcess := True; // Lock the input data GUI - user no more can edit it
  for s in urlList do
    pipeline.Input.Add(s);
  pipeline.Input.CompleteAdding;
end;

procedure TfrmMain.FlushData;
var v: TOmniValue;
begin
  if pipeline = nil then exit;
  if pipeline.Output = nil then exit;
  if pipeline.Output.IsFinalized then
  begin
    InProcess := False;  
    exit;
  end;

  Memo2.Lines.BeginUpdate;
  try
    while pipeline.Output.TryTake(v) do
      Memo2.Lines.Add( v.AsString );
  finally
    Memo2.Lines.EndUpdate;
  end;

  // optionally - scroll output memo2 to the last line 
end;

注意一些细节，思考它们并理解它们的本质：

1. 只有 FlushData 正在更新输出备忘录。 FlushData 从TTimer 事件或表单模式属性设置器中调用。它们都只能从主 VCL 线程调用。因此，FlushData 永远不会被称为后台线程。

2. Retriever 是一个免费的独立函数，它不是表单的成员，它对表单一无所知，也不会引用您的表单实例。这样您就可以实现两个目标：避免“紧密耦合”，避免从后台线程错误地访问表单控件的机会，这在 VCL 中是不允许的。 检索器函数在后台线程中工作，它们确实加载数据、存储数据，但它们从不接触 GUI。就是这个想法。

经验法则 - 表单的所有方法都只能从主 VCL 线程调用。所有流水线阶段子例程——后台线程的主体——都被声明并在任何 VCL 表单之外工作，并且无法访问任何一个。这些领域之间不应混杂。

1. 您将 GUI 更新限制为固定刷新率。而且这个速度不应该太频繁。 Windows GUI 和用户的眼睛应该有时间赶上。

2. 您的表单以两种明确界定的模式运行 - InProcess 和 not InProcess。在这些模式中，用户可以使用不同的功能和控件集。它还管理模式到模式的转换，例如清除输出备忘录文本、提醒用户状态更改、释放已用线程管理对象的内存（此处：管道）等。因此，仅更改此属性（调用 setter）来自主 VCL 线程，从不来自后台工作人员。 ＃2也有帮助。

1234563下一个计时器 olling 事件。这可能是管道了解有关表单并对其进行任何引用的唯一地方。但这打开了 Windows 消息传递、HWND 娱乐和其他我不想放在这里的微妙事物的罐头。