如何在 Delphi 中使用管道模式

Question

我正在尝试在我的测试项目 () 中实施流水线模式，但在将 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

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

一般来说，您有点过度使用了管道。

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

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

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

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

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

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

您应该将此表单设置为双模式，以便在下载运行ning 和非下载时具有不同的启用控件集。我通常使用特殊的布尔值属性来完成它，就像我在您链接的主题中向您展示的那样。您的用户不应该在管道进行时更改列表和设置（除非您将实施实时任务更改，但您还没有）。当从工作模式切换到空闲模式时，此模式切换器也是释放已完成管道对象的好地方。
如果你想玩管道工人链接，那么你可以把输入集合而不是 URL 字符串本身，而是那些的数组 - Memo1.Lines.ToArray()，那么您可以从 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;

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

只有FlushData正在更新输出备忘录。从 TTimer 事件或从表单模式属性 setter 调用 FlushData。它们都只能从主 VCL 线程调用。因此 FlushData 永远不会从后台线程中调用。
Retriever 是一个免费的独立函数，它不是表单的成员，它对表单一无所知，也没有引用您的表单实例。这样你就实现了两个目标：你避免了 "tight coupling" 并且你避免了从后台线程错误地访问表单控件的机会，这在 VCL 中是不允许的。检索器函数在后台线程中工作，它们加载数据，存储数据，但它们从不接触 GUI。就是这个意思。

经验法则 - 窗体的所有方法仅从主 VCL 线程调用。所有管道阶段子例程 - 后台线程的主体 - 都在任何 VCL 表单之外声明和工作，并且无法访问其中的 none。这些领域之间不应混合。

您将 GUI 更新限制为固定的刷新率。而且这个频率不应该太频繁。 WindowsGUI和用户眼睛应该有时间赶上。
您的表单以两种明确划分的模式运行 - InProcess 和 not InProcess。在这些模式中，用户可以使用不同的功能和控件集。它还管理模式到模式的转换，例如清除输出备忘录文本、提醒用户状态更改、释放已用线程管理对象（此处为：管道）的内存等。因此，仅此属性被更改（ setter 被调用）来自主 VCL 线程，从不来自后台工作者。 #2 也对此有所帮助。
未来可能的增强功能是使用 pipeline.OnStop 事件向您的表单发出带有自定义 Windows 消息的 PostMessage，因此它会切换工作完成后立即进入模式，而不是等待下一个计时器 olling 事件。这可能是管道对表单有任何了解并对其有任何引用的唯一地方。但这打开了 Windows 消息、HWND 娱乐和其他我不想放在这里的微妙的东西。

如何在 Delphi 中使用管道模式

How to use Pipeline pattern in Delphi

delphi

multithreading

http

omnithreadlibrary

delphi-xe8