登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 资源区 › 代码 ›mormot.core.threads--TSynThreadPool

mormot.core.threads--TSynThreadPool

晁红叶 3 天前
mormot.core.threads--TSynThreadPool
  1. { ************ 面向服务器进程的线程池 }  
  3. TSynThreadPool = class; // 前向声明TSynThreadPool类
  5. /// 定义了TSynThreadPool所使用的工作线程
  6. TSynThreadPoolWorkThread = class(TSynThread)
  7. protected
  8.     fOwner: TSynThreadPool; // 线程池所有者
  9.     fThreadNumber: integer; // 线程编号
  10.     {$ifndef USE_WINIOCP} // 如果不使用Windows I/O完成端口
  11.     fProcessingContext: pointer; // 正在处理的上下文，受fOwner.fSafe.Lock保护
  12.     fEvent: TSynEvent; // 同步事件
  13.     {$endif USE_WINIOCP}
  14.     procedure NotifyThreadStart(Sender: TSynThread); // 通知线程开始
  15.     procedure DoTask(Context: pointer); // 异常安全地调用fOwner.Task()
  16. public
  17.     /// 初始化线程
  18.     constructor Create(Owner: TSynThreadPool); reintroduce;
  19.   
  20.     /// 终结线程
  21.     destructor Destroy; override;
  22.   
  23.     /// 循环等待并执行挂起的任务，通过调用fOwner.Task()
  24.     procedure Execute; override;
  25.   
  26.     /// 关联的线程池
  27.     property Owner: TSynThreadPool
  28.       read fOwner;
  29. end;
  31. TSynThreadPoolWorkThreads = array of TSynThreadPoolWorkThread; // 线程池工作线程数组类型
  33. /// 一个简单的线程池，用于例如快速处理HTTP/1.0请求
  34. // - 在Windows上通过I/O完成端口实现，或在Linux/POSIX上使用经典的事件驱动方法
  35. TSynThreadPool = class
  36. protected
  37.     {$ifndef USE_WINIOCP} // 如果不使用Windows I/O完成端口
  38.     fSafe: TOSLightLock; // 轻量化锁，但可能不是最稳定的
  39.     {$endif USE_WINIOCP}
  40.     fWorkThread: TSynThreadPoolWorkThreads; // 工作线程数组
  41.     fWorkThreadCount: integer; // 工作线程数量
  42.     fRunningThreads: integer; // 正在运行的线程数量
  43.     fExceptionsCount: integer; // 异常计数（未在代码中明确使用，但可能用于调试或监控）
  44.     fContentionAbortDelay: integer; // 由于争用而拒绝连接的延迟时间（毫秒）
  45.     fOnThreadTerminate: TOnNotifyThread; // 线程终止通知事件
  46.     fOnThreadStart: TOnNotifyThread; // 线程开始通知事件
  47.     fContentionTime: Int64; // 等待队列可用槽位的总时间（毫秒）
  48.     fContentionAbortCount: cardinal; // 由于争用而中止的任务数
  49.     fContentionCount: cardinal; // 等待队列可用槽位的次数
  50.     fName: RawUtf8; // 线程池名称
  51.     fTerminated: boolean; // 线程池是否已终止
  52.     {$ifdef USE_WINIOCP} // 如果使用Windows I/O完成端口
  53.     fRequestQueue: THandle; // IOCP有其自己的内部队列
  54.     {$else}
  55.     fQueuePendingContext: boolean; // 当所有线程都忙时，是否应维护一个内部队列
  56.     fPendingContext: array of pointer; // 挂起的上下文数组
  57.     fPendingContextCount: integer; // 挂起的上下文数量
  58.     function GetPendingContextCount: integer; // 获取挂起上下文数量的函数
  59.     function PopPendingContext: pointer; // 从挂起上下文数组中弹出一个元素的函数
  60.     function QueueLength: integer; virtual; // 获取队列长度的虚拟函数（可能用于调试）
  61.     {$endif USE_WINIOCP}
  62.     /// 在I/O错误时结束线程
  63.     function NeedStopOnIOError: boolean; virtual;
  64.     /// 在通知后要执行的进程，这是一个抽象方法，需要被子类实现
  65.     procedure Task(aCaller: TSynThreadPoolWorkThread; aContext: pointer); virtual; abstract;
  66.     /// 中止任务的进程
  67.     procedure TaskAbort(aContext: Pointer); virtual;
  69. public
  70.     /// 使用指定的线程数初始化线程池
  71.     // - 抽象的Task()方法将由其中一个线程调用
  72.     // - 一个线程池最多可以关联256个线程
  73.     // - 在Windows上，可以可选地接受一个之前使用Windows重叠I/O（IOCP）打开的aOverlapHandle
  74.     // - 在POSIX上，如果aQueuePendingContext=true，则将挂起的上下文存储到内部队列中，
  75.     //   以便在队列未满时Push()返回true
  76.     {$ifdef USE_WINIOCP}
  77.     constructor Create(NumberOfThreads: integer = 32; aOverlapHandle: THandle = INVALID_HANDLE_VALUE; const aName: RawUtf8 = '');
  78.     {$else}
  79.     constructor Create(NumberOfThreads: integer = 32; aQueuePendingContext: boolean = false; const aName: RawUtf8 = '');
  80.     {$endif USE_WINIOCP}
  81.   
  82.     /// 关闭线程池，释放所有关联的线程
  83.     destructor Destroy; override;
  84.   
  85.     /// 让线程池处理一个指定的任务（作为指针）
  86.     // - 如果没有空闲线程可用，并且Create(aQueuePendingContext=false)被使用，则返回false（调用者稍后应重试）
  87.     // - 如果在Create中aQueuePendingContext为true，或使用了IOCP，则提供的上下文将被添加到内部列表，并在可能时处理
  88.     // - 如果aWaitOnContention默认为false，则在队列满时立即返回
  89.     // - 设置aWaitOnContention=true以等待最多ContentionAbortDelay毫秒并重试将任务排队
  90.     function Push(aContext: pointer; aWaitOnContention: boolean = false): boolean;
  91.     {$ifndef USE_WINIOCP}
  92.   
  93.     /// 在Push()返回false后调用，以查看队列是否确实已满
  94.     // - 如果QueuePendingContext为false，则返回false
  95.     function QueueIsFull: boolean;
  96.   
  97.     /// 如果所有线程都忙时，线程池是否应维护一个内部队列
  98.     // - 作为Create构造函数的参数提供
  99.     property QueuePendingContext: boolean
  100.       read fQueuePendingContext;
  101.     {$endif USE_WINIOCP}
  102.   
  103.     /// 对此线程池中定义的线程的低级访问
  104.     property WorkThread: TSynThreadPoolWorkThreads
  105.       read fWorkThread;
  106. published
  107.     /// 线程池中可用的线程数
  108.     // - 映射Create()参数，即默认为32
  109.     property WorkThreadCount: integer
  110.       read fWorkThreadCount;
  111.   
  112.     /// 当前在此线程池中处理任务的线程数
  113.     // - 范围在0..WorkThreadCount之间
  114.     property RunningThreads: integer
  115.       read fRunningThreads;
  116.   
  117.     /// 由于线程池争用而被拒绝的任务数
  118.     // - 如果此数字很高，请考虑设置更高的线程数，或分析并调整Task方法
  119.     property ContentionAbortCount: cardinal
  120.       read fContentionAbortCount;
  121.   
  122.     /// 由于争用而拒绝连接的延迟时间（毫秒）
  123.     // - 默认为5000，即等待IOCP或aQueuePendingContext内部列表中有空间可用5秒
  124.     // - 在此延迟期间，不接受新的连接（即不调用Accept），以便负载均衡器可以检测到争用并切换到池中的另一个实例，
  125.     //   或直接客户端最终可能会拒绝其连接，因此不会开始发送数据
  126.     property ContentionAbortDelay: integer
  127.       read fContentionAbortDelay write fContentionAbortDelay;
  128.   
  129.     /// 等待队列中可用槽位的总时间（毫秒）
  130.     // - 争用不会立即失败，但会重试直到ContentionAbortDelay
  131.     // - 此处的高数值需要对Task方法进行代码重构
  132.     property ContentionTime: Int64
  133.       read fContentionTime;
  134.   
  135.     /// 线程池等待队列中可用槽位的次数
  136.     // - 争用不会立即失败，但会重试直到ContentionAbortDelay
  137.     // - 此处的高数值可能需要增加线程数
  138.     // - 使用此属性和ContentionTime来计算平均争用时间
  139.     property ContentionCount: cardinal
  140.       read fContentionCount;
  141.   
  142.     {$ifndef USE_WINIOCP}
  143.     /// 当前等待分配给线程的输入任务数
  144.     property PendingContextCount: integer
  145.       read GetPendingContextCount;
  146.     {$endif USE_WINIOCP}
  147. end;
  149. {$M-} // 关闭内存管理消息
  151. const
  152.   // 允许TSynThreadPoolWorkThread堆栈使用最多256 * 2MB = 512MB的RAM
  153.   THREADPOOL_MAXTHREADS = 256;
复制代码
由于 TSynThreadPool类是一个高度抽象且依赖于特定实现的类（如它可能使用Windows的I/O完成端口或Linux/POSIX的事件驱动机制），编写一个完整的例程代码可能会相当复杂，并且需要模拟或实际实现这些依赖项。然而，我可以提供一个简化的示例，该示例展示了如何创建 TSynThreadPool实例、如何向其推送任务，并如何大致实现 Task方法。
请注意，以下代码是一个高度简化的示例，并不包含所有 TSynThreadPool类定义中的功能，特别是与Windows I/O完成端口或Linux/POSIX事件驱动机制相关的部分。此外，由于 TSynThreadPool是一个假设的类（因为它不是Delphi标准库或广泛认可的第三方库的一部分），我将基于您提供的类定义来编写这个示例。
  1. program TSynThreadPoolDemo;
  3. {$MODE DELPHI}
  5. uses
  6.   SysUtils, Classes; // 引入必要的单元
  8. // 假设TSynThreadPool及其依赖项已经在某个单元中定义
  9. // 这里我们使用一个占位符单元名YourThreadPoolUnit
  10. uses YourThreadPoolUnit;
  12. // 一个简单的任务上下文类（仅作为示例）
  13. type
  14.   TMyTaskContext = record
  15.     Data: Integer;
  16.   end;
  18. // TSynThreadPool的Task方法的实现类
  19. type
  20.   TMyThreadPool = class(TSynThreadPool)
  21.   protected
  22.     procedure Task(aCaller: TSynThreadPoolWorkThread; aContext: Pointer); override;
  23.   end;
  25. { TMyThreadPool }
  27. procedure TMyThreadPool.Task(aCaller: TSynThreadPoolWorkThread; aContext: Pointer);
  28. var
  29.   Ctx: PMyTaskContext;
  30. begin
  31.   Ctx := PMyTaskContext(aContext);
  32.   WriteLn('Processing task with data: ', Ctx.Data);
  33.   // 在这里添加处理任务的代码
  34.   // ...
  35. end;
  37. var
  38.   Pool: TMyThreadPool;
  39.   Ctx: TMyTaskContext;
  40.   I: Integer;
  42. begin
  43.   try
  44.     // 创建一个线程池实例，假设我们想要使用4个工作线程
  45.     Pool := TMyThreadPool.Create(4);
  46.     try
  47.       // 模拟向线程池推送一些任务
  48.       for I := 1 to 10 do
  49.       begin
  50.         Ctx.Data := I;
  51.         if not Pool.Push(@Ctx) then
  52.         begin
  53.           // 在这个简化的示例中，我们不会处理Push返回false的情况
  54.           // 在实际应用中，您可能需要等待、重试或将任务放入另一个队列中
  55.           WriteLn('Failed to push task to the pool (this should not happen in this simplified example)');
  56.         end;
  57.       end;
  59.       // 在这个简化的示例中，我们没有等待所有任务完成
  60.       // 在实际应用中，您可能需要等待线程池中的所有任务都完成后再继续
  61.       // ...
  63.     finally
  64.       // 销毁线程池实例，这将释放所有关联的资源
  65.       Pool.Free;
  66.     end;
  67.   except
  68.     on E: Exception do
  69.       WriteLn('An error occurred: ', E.Message);
  70.   end;
  71.   // 保持控制台窗口打开，直到用户按任意键
  72.   WriteLn('Press Enter to exit...');
  73.   ReadLn;
  74. end.
  76. // 注意：由于TSynThreadPool是一个假设的类，并且上述代码没有实现所有细节（如线程池的实际工作线程管理、任务队列等），
  77. // 因此这个示例主要是为了展示如何使用该类（如果它存在的话）的大致结构。
  78. // 在实际应用中，您需要根据TSynThreadPool类的具体实现来调整此代码。
复制代码
重要说明：

  • 占位符单元：在上述代码中，我使用了 YourThreadPoolUnit作为包含 TSynThreadPool类定义的占位符单元名。在实际应用中，您需要将其替换为包含该类定义的实际单元名。
  • 任务上下文：我定义了一个简单的 TMyTaskContext记录类型来作为任务的上下文。在实际应用中，您可能需要根据需要定义更复杂的上下文类型。
  • 错误处理：在 Push方法返回 false的情况下，上述代码仅打印了一条消息，并没有采取任何恢复措施。在实际应用中，您可能需要实现更复杂的错误处理逻辑（如重试、等待或将任务放入另一个队列中）。
  • 等待任务完成：上述代码没有等待线程池中的所有任务都完成。在实际应用中，您可能需要实现某种形式的等待机制（例如，使用同步事件或计数器）来确保所有任务都已完成后再继续执行后续代码。
  • 线程池实现：由于 TSynThreadPool是一个假设的类，并且其实现细节（如工作线程的管理、任务队列的实现等）并未在您的类定义中给出，因此上述代码仅提供了一个大致的框架。在实际应用中，您需要根据 TSynThreadPool类的具体实现来调整此代码。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
相关推荐
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
晁红叶
3 天前

0

粉丝关注

7

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫