OmniThreadLibrary 背景工作執行緒模式

– 9 分鐘閱讀

現代應用程式越來越強調使用者經驗，避免主執行緒阻塞至關重要。OmniThreadLibrary 提供的 BackgroundWorker 模式讓 Delphi 開發者能輕鬆地將耗時操作移至背景執行緒，確保主執行緒的流暢性。藉由設定執行緒數量、初始化與終止函式，以及工作完成回呼，BackgroundWorker 能有效管理非同步任務。本文提供的程式碼範例示範如何使用 BackgroundWorker 下載網頁內容，並利用正規表示式解析關鍵資訊。同時也說明瞭如何在下載過程中更新 UI，提供使用者即時回饋，提升整體應用程式體驗。

procedure TForm1.btnStartClick(Sender: TObject);
begin
  if FBackgroundWorker = nil then
  begin
    FBackgroundWorker := Parallel.BackgroundWorker
      .NumTasks(4)
      .Initialize(
        procedure (const task: IOmniTask; var taskState: TOmniValue)
        begin
          taskState := THTTPClient.Create;
        end)
      .Finalize(
        procedure (const task: IOmniTask; const taskState: TOmniValue)
        begin
          taskState.AsObject.Free;
        end)
      .Execute(
        procedure (const task: IOmniTask; const workItem: IOmniWorkItem)
        var
          response: IHTTPResponse;
        begin
          response := workItem.TaskState.ToObject<THTTPClient>.Get(workItem.Data);
          if (response.StatusCode div 100) = 2 then
            workItem.Result := response.ContentAsString;
        end)
        .OnRequestDone(OnRequestDone)
      .Create;
  end;

  FBackgroundWorker.Schedule(
    FBackgroundWorker.CreateWorkItem('https://example.com'));
end;

探討OmniThreadLibrary中的BackgroundWorker模式

在現代軟體開發中，多執行緒程式設計已成為提升應用程式效能的關鍵技術之一。Delphi開發者可以利用OmniThreadLibrary（OTL）這個強大的平行程式函式庫來簡化多執行緒程式的開發。其中，BackgroundWorker模式是OTL提供的一個重要功能，能夠有效地將耗時的操作放到背景執行緒中執行，從而避免阻塞主執行緒，提升使用者經驗。

BackgroundWorker模式的基本設定

要使用BackgroundWorker模式，首先需要建立一個BackgroundWorker的例項。在OTL中，這可以透過Pattern.BackgroundWorker工廠函式來實作。以下是一個基本的設定範例：

FBackgroundWorker := Pattern.BackgroundWorker
  .NumTasks(4)
  .Initialize(InitializeWorkerTask_asy)
  .Finalize(FinalizeWorkerTask_asy)
  .Execute(DownloadWebPage_asy)
  .OnRequestDone(OnRequestDone)
  .Create;

內容解密：

  • .NumTasks(4) 指定了背景執行緒的數量為4，可以根據實際需求進行調整。
  • .Initialize(InitializeWorkerTask_asy) 和 .Finalize(FinalizeWorkerTask_asy) 分別指定了背景執行緒初始化和終止時要執行的函式。
  • .Execute(DownloadWebPage_asy) 指定了背景執行緒中要執行的任務。
  • .OnRequestDone(OnRequestDone) 指定了當任務完成後在主執行緒中要執行的函式。

背景執行緒的初始化與終止

在上述範例中，InitializeWorkerTask_asy用於初始化背景執行緒的狀態，而FinalizeWorkerTask_asy則負責清理在初始化階段建立的資源。以下是這兩個函式的範例實作：

procedure TfrmBackgroundWorker.InitializeWorkerTask_asy(var taskState: TOmniValue);
begin
  taskState := THTTPClient.Create;
end;

procedure TfrmBackgroundWorker.FinalizeWorkerTask_asy(const taskState: TOmniValue);
begin
  taskState.AsObject.Free;
end;

內容解密：

  • 在InitializeWorkerTask_asy中，為每個背景執行緒建立了一個THTTPClient例項，用於後續的網頁下載操作。
  • 在FinalizeWorkerTask_asy中，釋放了在初始化階段建立的THTTPClient例項，避免記憶體洩漏。

提交工作請求

當使用者點選搜尋按鈕時，應用程式會提交一個新的工作請求到BackgroundWorker。以下是一個提交工作請求的範例：

procedure TfrmBackgroundWorker.btnSearchClick(Sender: TObject);
begin
  FBackgroundWorker.Schedule(
    FBackgroundWorker.CreateWorkItem('https://en.delphipraxis.net/search/?q=' + inpSearch.Text),
    FBackgroundWorker.Config.OnRequestDone(ListDownloaded));
end;

內容解密：

  • CreateWorkItem用於建立一個新的工作請求，請求的資料是一個包含搜尋字串的URL。
  • Schedule方法將工作請求提交到BackgroundWorker中執行，並指定了一個自訂的完成回呼函式ListDownloaded。

背景工作執行緒（Background Worker）實戰解析

在現代軟體開發中，背景工作執行緒扮演著提升應用程式效能與使用者經驗的關鍵角色。本文將探討如何利用背景工作執行緒實作高效的非同步網頁下載與處理。

網頁下載的非同步實作

在處理網路請求時，傳統的同步方法往往會阻塞主執行緒，導致介面凍結。以下程式碼展示瞭如何利用 IOmniBackgroundWorker 實作非同步網頁下載：

response := workItem.TaskState.ToObject<THTTPClient>.Get(workItem.Data);
if (response.StatusCode div 100) = 2 then
    workItem.Result := response.ContentAsString;

內容解密：

  1. workItem.TaskState.ToObject<THTTPClient> 將儲存的 HTTP 客戶端物件取出，用於傳送網路請求。
  2. response.StatusCode div 100 判斷 HTTP 狀態碼的第一位數字，用於檢查請求是否成功（2xx 狀態碼）。
  3. 若成功，將網頁內容儲存於 workItem.Result 中，供主執行緒後續處理。

下載完成的後續處理

當背景工作執行緒完成網頁下載後，主執行緒會接收到結果並進行處理：

procedure TfrmBackgroundWorker.ListDownloaded(
    const Sender: IOmniBackgroundWorker;
    const workItem: IOmniWorkItem);
var
    filter, url: string;
    hrefMatch: TRegEx;
    match: TMatch;
    index: integer;
    request: IOmniWorkItem;
begin
    if workItem.Result.IsEmpty then
        ShowMessage('背景工作執行緒下載失敗: ' + workItem.Data)
    else if Assigned(FBackgroundWorker) then begin
        // 正規表示式解析與後續處理...
    end;
end;

內容解密：

  1. 檢查 workItem.Result 是否為空，以判斷下載是否成功。
  2. 若下載失敗，顯示錯誤訊息；若成功，則進行 HTML 解析。
  3. 使用正規表示式提取搜尋結果中的連結與標題。
  4. 為每個提取的連結建立新的下載請求，並排程至背景工作執行緒。

文章下載與狀態更新

當單篇文章下載完成後，會觸發 ArticleDownloaded 方法：

procedure TfrmBackgroundWorker.ArticleDownloaded(
    const Sender: IOmniBackgroundWorker;
    const workItem: IOmniWorkItem);
var
    request: TRequest;
begin
    if not FRequests.TryGetValue(workItem.UniqueID, request) then
        Exit;
    request.Page := workItem.Result;
    FRequests.AddOrSetValue(workItem.UniqueID, request);
    chkSitelist.Checked[request.Index] := True;
end;

內容解密：

  1. 使用 workItem.UniqueID 從 FRequests 字典中檢索對應的請求資料。
  2. 將下載的網頁內容儲存於 request.Page 中。
  3. 更新 UI，在清單中標記已完成的下載專案。

使用者互動處理

當使用者點選清單專案時，會觸發 chkSitelistClick 事件：

procedure TfrmBackgroundWorker.chkSitelistClick(Sender: TObject);
var
    request: TRequest;
begin
    if FRequests.TryGetValue(FTitles[chkSiteList.ItemIndex], request) then
        HtmlViewer.LoadFromString(request.Page)
    else
        HtmlViewer.Clear;
end;

內容解密：

  1. 將清單索引對應至唯一的請求 ID。
  2. 若該請求已完成下載，則在 HTML 檢視器中顯示對應內容。

平行處理模式進階探討

在前面的章節中，我們已經深入瞭解了平行處理的基本概念以及使用標準PPL工具實作管線（Pipeline）的相關技術。雖然這種實作方式很有用，但實際操作起來卻略顯笨拙。本章節將重點介紹OTL函式庫中的管線實作，並透過ParallelPipeline範例程式進行詳細解析。

管線模式的應用場景

在開始探討OTL的管線實作之前，讓我們先回顧一下管線的基本概念。管線模式適用於將資料處理過程拆分成多個獨立的階段，並在多個背景執行緒中平行執行這些階段。ParallelPipeline範例程式示範了一個簡單的檔案分析器，它能夠掃描指定資料夾（包含子資料夾）中所有的.pas副檔名檔案，統計每個檔案的行數、單字數和字元數，並顯示所有找到的檔案的累積統計資訊。

實作需求分析

從實作的角度來看，這個程式需要完成以下任務：

  • 找出資料夾和子資料夾中的所有檔案
  • 將每個檔案從磁碟讀入緩衝區
  • 統計緩衝區中的行數、單字數和字元數
  • 更新累積統計資訊

管線設計思路

為了實作上述功能，我們可以將整個處理過程設計成一系列的工作階段，並透過通訊通道將這些階段連線起來。這種設計思路實際上是在改變演算法，以達到提升程式效能的目的。

在這個例子中，第一個階段負責列舉檔案，它的輸入是待掃描的資料夾名稱，輸出則是找到的所有.pas檔案名稱。與其將找到的檔案名稱儲存在某個列表中，不如直接將它們寫入通訊通道。

管線階段的詳細設計

第二個階段是一個簡單的檔案讀取器，它透過共用的通訊通道與第一階段相連。該階段從通道中讀取下一個檔案名稱，開啟檔案並將其內容讀入緩衝區，然後將該緩衝區寫入自己的輸出通道。

這種階段與通道交替連線的模式持續下去。第三階段負責分析資料，它從輸入通道（同時也是檔案讀取階段的輸出）讀取資料緩衝區，對資料進行分析，並將分析結果寫入自己的輸出通道。

第四個也是最後一個階段負責匯總所有的區域性分析結果，並生成最終的統計報告，主程式在作業完成後存取該報告。

管線實作的優點

這種設計方式允許我們建立一系列小型的工作單元，並透過通訊通道將它們連線起來。工作單元之間的獨立性使得平行化處理變得更加簡單。OTL函式庫使用IOmniBlockingCollection來實作管線模式中的通訊通道。

建立管線

接下來，讓我們來看看具體的程式碼實作。使用者介面上有一個按鈕，可以啟動或停止平行作業。當啟動處理程式時，它會呼叫CreatePipeline方法。要停止處理程式，只需呼叫管線的Cancel方法。

procedure TfrmPipeline.btnStartClick(Sender: TObject);
begin
  if btnStart.Tag = 0 then begin
    ListBox1.Items.Add(inpStartFolder.Text);
    CreatePipeline;
    btnStart.Caption := 'Stop';
    btnStart.Tag := 1;
    TimerUpdateProcessing.Enabled := true;
  end
  else
    if assigned(FPipeline) then
      FPipeline.Cancel;
end;

CreatePipeline 方法詳解

procedure TfrmPipeline.CreatePipeline;
begin
  FPipeline := Parallel.Pipeline;
  FPipeline.Stage(FolderScanner);
  FPipeline.Stage(FileReader,
    Parallel.TaskConfig.OnMessage(
      HandleFileReaderMessage))
    .NumTasks(Environment.Process.Affinity.Count)
    .Throttle(1000);
  FPipeline.Stage(StatCollector)
    .NumTasks(2);
  FPipeline.Stage(StatAggregator);
  FPipeline.OnStopInvoke(ShowResult);
  FPipeline.Run;
  FPipeline.Input.Add(inpStartFolder.Text);
  FPipeline.Input.CompleteAdding;
end;

內容解密：

  1. Parallel.Pipeline 初始化：建立一個新的管線例項。
  2. FPipeline.Stage：新增不同的處理階段到管線中。
    • FolderScanner：第一階段，負責掃描資料夾。
    • FileReader：第二階段，負責讀取檔案內容。使用NumTasks設定多執行緒數量，並使用Throttle控制輸入速率。
    • StatCollector：第三階段，負責統計檔案資訊。使用NumTasks(2)設定兩個平行任務。
    • StatAggregator：第四階段，負責匯總統計結果。
  3. OnStopInvoke(ShowResult)：當管線停止時，呼叫ShowResult方法顯示結果。
  4. FPipeline.Run：啟動管線。
  5. FPipeline.Input.Add(inpStartFolder.Text)：將初始資料夾路徑加入管線輸入。
  6. FPipeline.Input.CompleteAdding：標記輸入完成，不再接受新的輸入。

此段程式碼展示瞭如何使用OTL函式庫建立和管理一個管線處理流程，實作複雜任務的平行處理和效能最佳化。

管道處理（Pipeline）實作詳解

在現代軟體開發中，平行處理已成為提升程式效能的重要手段。本文將探討如何使用 IOmniPipeline 實作一個高效的管道處理機制，並詳細分析其內部實作原理與最佳實踐。

管道建立與組態

首先，我們需要建立一個 IOmniPipeline 物件，並透過 Parallel.Pipeline 方法初始化管道。接著，透過 Stage 方法建立多個處理階段，每個階段負責不同的任務。

FPipeline := Parallel.Pipeline;
FPipeline.Stage(FolderScanner)
           .Stage(FileReader)
           .Stage(StatCollector)
           .Stage(StatAggregator);

內容解密：

  1. FPipeline 初始化：使用 Parallel.Pipeline 建立管道物件。
  2. Stage 建立：透過 Stage 方法加入多個處理階段，包括資料夾掃描、檔案讀取、統計收集與彙總。
  3. 流暢介面（Fluent Interface）：使用 OTL 提供的流暢介面風格進行組態，使程式碼更具可讀性。

各階段詳細組態

在建立階段後，我們需要對每個階段進行詳細組態。例如，第二階段（FileReader）需要設定平行任務數量、輸出佇列大小以及訊息處理函式。

.Stage(FileReader)
.NumTasks(NumCPUs)
.Throttle(1000)
.TaskConfig(OnMessage);

內容解密：

  1. NumTasks(NumCPUs)：根據 CPU 核心數設定平行任務數量，以充分利用硬體資源。
  2. Throttle(1000)：限制輸出佇列大小為 1000，防止記憶體過度消耗。
  3. TaskConfig(OnMessage)：設定訊息處理函式，將階段輸出傳送至主執行緒進行處理。

管道啟動與執行

完成組態後，呼叫 Run 方法啟動管道。接著，將初始資料夾寫入輸入通道以啟動整個處理流程。

FPipeline.Run;
FPipeline.Input.Add(起始資料夾);
FPipeline.CompleteAdding;

內容解密：

  1. FPipeline.Run：啟動管道，開始執行所有階段的任務。
  2. FPipeline.Input.Add：將初始資料夾加入輸入通道，啟動資料處理流程。
  3. FPipeline.CompleteAdding：通知第一階段不再有新輸入，使其在完成後正確終止。

通訊通道大小與節流（Throttling）機制

預設情況下，每個通訊通道可儲存最多 10,240 個元素。當達到此限制時，生產者任務將進入睡眠狀態，直到通道部分清空。

.Throttle(1000)

內容解密：

  1. 節流機制：限制輸出佇列大小，防止快速生產者填滿記憶體。
  2. 動態調整：在實際應用中，需要根據不同場景測試並調整通道大小和任務數量，以獲得最佳效能。

第一階段實作：資料夾掃描（FolderScanner）

第一階段負責掃描指定資料夾，並將找到的 Pascal 檔案路徑輸出至下一階段。

procedure TfrmPipeline.FolderScanner(
  const input, output: IOmniBlockingCollection;
  const task: IOmniTask);
var
  value: TOmniValue;
begin
  for value in input do begin
    DSiEnumFilesEx(
      IncludeTrailingPathDelimiter(value) + '*.pas',
      0, true,
      procedure (const folder: string; S: TSearchRec; isAFolder: boolean; var stopEnum: boolean)
      begin
        stopEnum := task.CancellationToken.IsSignalled;
        if (not stopEnum) and (not isAFolder) then
          output.TryAdd(IncludeTrailingPathDelimiter(folder) + S.Name);
      end);
  end;
end;

內容解密：

  1. DSiEnumFilesEx：使用第三方函式庫遞迴列舉指定資料夾中的 .pas 檔案。
  2. output.TryAdd：將符合條件的檔案路徑加入輸出通道，供下一階段處理。
  3. task.CancellationToken.IsSignalled：檢查是否收到取消訊號，以支援優雅地終止操作。
玄貓 BlackCat

玄貓 BlackCat

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。