multithreading - 如何在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

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

作为一般性说明，您会过度使用管道。

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

换句话说，Inserter并不是您所需要的。您在管道中需要的只是其Input集合，一个下载程序过程和Output集合。是的，对于复杂的事物管道来说，这是非常简单的任务，这就是为什么我之前提到了另外两个更简单的模式。

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

您不应该等待管道在VCL主线程中完成。相反，您应该拆开pipeleine，让其完全在后台运行。将对创建的管道的引用保存到Form的member变量中，以便您可以从TTimer事件访问其Output集合，还可以在其过程结束后释放管道。

您应该保持该变量链接到管道对象，直到下载结束为止，然后在此之后而不是之前将其设置为nil(释放对象)。您了解Delphi中的接口(interface)和引用计数，对吗？

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

您应该将此表单设为双模式，以便在运行下载和不运行下载时都具有一组不同的已启用控件。我通常使用特殊的 bool 属性来实现它，就像我在链接的主题中向您展示的那样。
在管道进行过程中，您的用户不应更改列表和设置(除非您要实现实时任务更改，但您尚未执行)。当从工作模式切换到空闲模式时，此模式切换器也是释放完成的管道对象的好地方。

如果您想使用管道工作程序链，则可以将URL字符串本身而不是URL字符串本身(而不是URL字符串)放入Input Collection中，然后从Unpacker阶段开始，该阶段从输入中获取字符串数组集合(实际上只有一个)并进行枚举，然后将字符串放入stage-output集合中。
但是，这几乎没有实用值(value)，因为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;

请注意一些细节，仔细考虑并了解其中的本质:

仅Memo1.Lines.ToArray()正在更新输出备忘。从FlushData事件或表单模式属性 setter 调用FlushData。两者都只能从VCL主线程调用。因此，TTimer永远不会称为表单后台线程。

FlushData是一个免费的独立函数，它不是表单的成员，并且对表单一无所知，也没有引用表单实例。这样就可以实现两个目标:避免“紧密耦合”，并且避免从后台线程错误地访问表单控件的可能性，这在VCL中是不允许的。
检索器功能在后台线程中工作，它们确实加载数据，确实存储数据，但是它们从未接触过GUI。那是主意。

经验法则-仅从主VCL线程调用表单中的所有方法。所有的管道阶段子例程(后台线程的主体)都被声明并且在任何VCL表单之外运行，并且无法访问这些子例程。这些领域之间不应混为一谈。

可以将GUI更新限制为固定的刷新率。而且该比率不应太频繁。 Windows GUI和用户的眼睛应该有时间 catch 。

您的表单以两种清晰描述的模式运行-Retriever和InProcess。在那些模式下，用户可以使用不同的功能和控件集。它还管理模式到模式的转换，例如清除输出注释文本，向用户发出状态更改警报，释放已使用的线程管理对象的内存(此处为管道)等。因此，仅更改了此属性(称为seter)来自VCL主线程，而不来自后台工作人员。 ＃2对此也有帮助。

将来可能的增强功能是使用not InProcess事件向您的表单中发送带有自定义Windows消息的pipeline.OnStop，因此它将在工作完成后立即切换模式，而无需等待下一个计时器olling事件。这可能是管道唯一了解表单并具有任何引用的地方。但这打开了Windows消息传递，HWND休闲和其他我不想放在这里的微妙功能的功能。