使用 Delphi 10.4 和 OmniThreadLibrary 高效处理大量 API 调用

Question

我正在寻找有关如何使用 Delphi 10.4 和 OmniThreadLibrary 有效处理大量 API 调用的指南。

具体来说，我需要对服务器进行大约 3,000 次 API 调用。为了优化性能，我想将并发线程数限制为 5。一旦一个线程处理完一个调用，我想启动另一个线程，直到所有 3,000 个调用都完成。

有人可以提供一个基本示例或提供有关如何使用 Delphi 和 OmniThreadLibrary 有效完成此任务的建议吗？谢谢您的协助！

我尝试利用调度机制，但没有提供令人满意的解决方案

Answer 1

因此，这些调用不是您需要一次性处理的一批；而是如果它是一个网络服务器，它们就会像它们那样进来，你需要查看将它们作为队列处理的任务。

第1步：定义任务队列和工作池 首先，为您的 API 调用任务定义一个线程安全队列。 OmniThreadLibrary 提供了可用于此目的的各种同步原语。然后，初始化一个工作任务池来处理该队列中的项目。

uses
  OtlTaskControl,
  OtlCollections,
  OtlCommon,
  OtlParallel;

type
  TApiTask = record
    // Define the API call parameters or any relevant information
    URL: string;
    // Add other fields as necessary
  end;

var
  TaskQueue: IOmniBlockingCollection;
  MaxThreads: Integer = 5;

第 2 步：创建处理 API 调用的过程 这是您实际 API 调用逻辑的占位符，因为您没有共享您喜欢做的事情。将其替换为执行 API 请求的实际代码。

procedure ProcessApiCall(const TaskDetail: TApiTask);
begin
  // Implement the API call logic here.
  // For example, HTTP GET/POST operations.
end;

第3步：初始化Worker并管理任务池 设置工人任务。每个工作人员重复从队列中取出一个任务并处理它。当队列为空时，工作人员可以等待新任务，或者在所有任务完成后终止。

procedure InitializeWorkerPool;
var
  i: Integer;
  worker: IOmniBackgroundWorker;
begin
  TaskQueue := TOmniBlockingCollection.Create;

  for i := 1 to MaxThreads do
  begin
    worker := CreateTask(
      procedure (const Task: IOmniTask)
      var
        ApiTask: TApiTask;
      begin
        while TaskQueue.TryTake(ApiTask) do
        begin
          ProcessApiCall(ApiTask);
        end;
      end)
      .Unobserved
      .Run;
  end;
end;

对于使用 Delphi 和 OmniThreadLibrary 的动态请求处理系统，其中任务在到达时排队并由有限数量的线程处理，这里有一个分步实现。此示例假设您对如何设置和使用 OmniThreadLibrary 有基本了解。

第1步：定义任务队列和工作池 首先，为您的 API 调用任务定义一个线程安全队列。 OmniThreadLibrary 提供了可用于此目的的各种同步原语。然后，初始化一个工作任务池，这些任务将处理该队列中的项目。

德尔福 复制代码 用途 OtlTaskControl, Otl收藏， OtlCommon, Otl并行；

类型 TApiTask = 记录 // 定义API调用参数或任何相关信息 网址：字符串； // 根据需要添加其他字段 结束；

var 任务队列：IOmniBlockingCollection； 最大线程数：整数 = 5； 第 2 步：创建处理 API 调用的过程 这是实际 API 调用逻辑的占位符。将其替换为执行 API 请求的实际代码。

德尔福 复制代码 过程 ProcessApiCall(const TaskDetail: TApiTask); 开始 // 在这里实现API调用逻辑。 // 例如，HTTP GET/POST 操作。 结尾; 第三步：初始化Worker并管理任务池 设置工人任务。每个工作人员重复从队列中获取任务并处理它们。当队列为空时，工作人员可以等待新任务，或者在所有任务完成后终止。

德尔福 复制代码 过程InitializeWorkerPool； 变量 i：整数； 工作人员：IOmniBackgroundWorker； 开始 TaskQueue := TOmniBlockingCollection.Create;

对于 i := 1 到 MaxThreads 执行 开始 工人 := 创建任务( 过程（常量任务：IOmniTask） 变量 ApiTask：TApiTask； 开始 而 TaskQueue.TryTake(ApiTask) 做 开始 ProcessApiCall(ApiTask); 结尾; 结尾） .未观察到的 。跑步; 结尾; 结束；

第4步：将任务添加到队列中 当您的应用程序收到 API 调用请求时，将它们添加到队列中。工作池将根据工作人员的可用性来管理执行。

procedure AddApiCallTask(const ApiTask: TApiTask);
begin
  TaskQueue.Add(ApiTask);
end;

第 5 步：完成 当您的应用程序即将关闭或需要正常停止处理时，请确保发出队列结束信号并等待所有任务完成。我会使用 bool 来表示传入的函数“它们”将在 .Net 中被拒绝，我们声明我们正在处置。

如何使用 您可能知道，但是要对请求进行排队

var
  ApiTask: TApiTask;
begin
  ApiTask.URL := 'http://example.com/api/';
  // Set other fields as needed.
  AddApiCallTask(ApiTask);
end;

使用我的示例，您需要在应用程序启动或开始处理时调用 InitializeWorkerPool，并在应用程序关闭或停止任务处理时调用 FinalizeTasks。

我总是对内存很偏执，尤其是在运行服务器应用程序时，所以我研究了使用 OmniThreadLibrary 的任务。这些是由Delphi内存管理系统自动处理的托管对象（基于接口和引用计数）；这就是理论。我喜欢重复使用东西，只是因为创建和释放内存很昂贵，但这只是我和处理服务器应用程序上的内存碎片。