threadpool.pas

ThreadPro 是本人开发的一套用于多线程编程的 Delphi 基础类库

{*******************************************************}
{                                                       }
{       高级线程池                                      }
{       单元名 ThreadPool.pas                           }
{                                                       }
{       版权所有 (C) 2006-6-25 飞翔的久久酷             }
{       http://www.flying99koo.com                      }
{       QQ:78273318 Mail: [e-mail]         }
{                                                       }
{*******************************************************}


unit ThreadPool;

interface

uses
  Windows, SysUtils, Classes,
  ThreadTask, HTTPtask, ComCtrls, StdCtrls;

type
  TThreadClass = class of TThreadTask;

  PThreadObj = ^TThreadObj;
  TThreadObj = record
    Handle : THandle;
    Active : Boolean;
    Index : Integer;
    TIndex : Integer;
    Success : Boolean;
  end;
  {<<<< Added 2006.6.22 at Company<<<<}
  PTaskObj = ^TTaskObj;
  TTaskObj = record
    Index : Integer;
    Param : Pointer;
  end;

  TRunOrder = (roNormal, roRandom);
  TRunMode = (rmNormal, rmSuddenDeath);

  TThreadPool = class
  private
    _TASK_OVER : Boolean;
    _THREAD_OVER : Boolean;
    _STOP_FLAG : Boolean;  //
  protected
    _THREADS : array of TThreadTask;
    _THREAD_LIST : TThreadList;
    procedure TaskDone(Sender : TObject);
    procedure SetThread(ThreadIndex : Integer; Task : Pointer); virtual; abstract;
    procedure CreateThreads;
    procedure CreateThread(ThreadIndex : Integer ; Task : Pointer); virtual;
    procedure RunThread(ThreadIndex : Integer);
    procedure TraceLog(Desc : string);
  public
    _THREAD_CLASS : TThreadClass;
    _THREAD_COUNT : Integer;
    _TASK_LIST : TThreadList;
    _RUN_ORDER : TRunOrder;
    _RUN_MODE : TRunMode;
    _SUCCESS_COUNT : Integer;  //for Normal RunMode
    _SUCCESS_INDEX : Integer;  //for SuddenDeath RunMode
    //
    _OnForceStop : TNotifyEvent;
    _OnTerminate : TNotifyEvent;
    _OnTraceLog : TTraceLogNotifyEvent;
    //
    _PROGRESS_BAR : TProgressBar;                //
    _TERM_PRO_BAR : TProgressBar;
    procedure Start;
    procedure Pause;
    procedure Goon;
    procedure Stop;
    constructor Create;
    destructor Destroy; override;
  end;

  THTTPtaskPool = class(TThreadPool)
  protected
    procedure SetThread(ThreadIndex : Integer; Task : Pointer);override;
  public
    P_READ_TIME_OUT : Integer;
    P_CONN_TIME_OUT : Integer;
    //
    P_USE_PROXY : boolean;
    P_PROXY_ADDR : string;
    P_PROXY_PORT : Integer;
    P_NEED_AUTH : boolean;
    P_PROXY_USER : string;
    P_PROXY_PASS : string;
    //
    P_HTTPtaskTrace : TTraceLogNotifyEvent;
  end;

  procedure LoadTask(Index : Integer; Param : Pointer; TaskList : TThreadList);

implementation

uses
  HTTPutil;


{TThreadPool}
constructor TThreadPool.Create;
begin
  inherited Create;
  _TASK_LIST := TThreadList.Create;
  _THREAD_LIST := TThreadList.Create;
  _THREAD_COUNT := 1;
  _TASK_OVER := False;
  _THREAD_OVER := False;
  _STOP_FLAG := False;
  _RUN_ORDER := roNormal;
  _RUN_MODE := rmNormal;
  _SUCCESS_COUNT := 0;
  _SUCCESS_INDEX := -1;
end;

destructor TThreadPool.Destroy;
var
  i : Integer;
begin
  with _TASK_LIST.LockList do
  try
    for i:=Count-1 downto 0 do
    begin
      FreeMem(PTaskObj(Items[i])^.Param);
      FreeMem(Items[i]);
      Delete(i);
    end;
  finally
    _TASK_LIST.UnlockList;
    _TASK_LIST.Free;
  end;

  with _THREAD_LIST.LockList do
  try
    for i:=Count-1 downto 0 do
    begin
      FreeMem(Items[i]);
      Delete(i);
    end;
  finally
    _THREAD_LIST.UnlockList;
    _THREAD_LIST.Free;
  end;
  
  inherited Destroy;
end;

procedure TThreadPool.TaskDone(Sender : TObject);
var
  i,idx : Integer;
  P : PThreadObj;
begin
  idx := -1;
  with _THREAD_LIST.LockList do
  try
    for i:=0 to Count-1 do
    begin
      P := PThreadObj(Items[i]);
      if not P^.Active then
      begin
        idx := P^.Index;
        if P^.Success then
        case _RUN_MODE of
        rmNormal : Inc(_SUCCESS_COUNT);
        rmSuddenDeath : begin
                          _STOP_FLAG := True;
                          _SUCCESS_INDEX := P^.TIndex;
                        end;
        end;
        FreeMem(Items[i]);
        Delete(i);
        if _STOP_FLAG or _TASK_OVER then
        begin
          if Count>0 then
            TraceLog(Format('还有 %d 个线程在运行',[Count]))
          else
            TraceLog('所有线程已安全终止...');
          if _TERM_PRO_BAR <> nil then
          _TERM_PRO_BAR.StepIt;;
        end;
        if Count=0 then
        _THREAD_OVER := True;
        _THREADS[idx] := nil;
        if _PROGRESS_BAR <> nil then
        _PROGRESS_BAR.StepIt;
        Break;
      end;
    end;
  finally
    _THREAD_LIST.UnlockList;
  end;
  if _THREAD_OVER then
  begin
    if not _STOP_FLAG then
    begin
      if _TASK_OVER then
        if Assigned(_OnTerminate) then
        _OnTerminate(Self);
    end else begin
      if Assigned(_OnForceStop) then
      _OnForceStop(Self);
    end;
  end else if (idx>=0)and(not _STOP_FLAG) then
    RunThread(idx);
end;

procedure TThreadPool.TraceLog(Desc : string);
begin
  if Assigned(_OnTraceLog) then
  _OnTraceLog(Desc,Self);
end;

procedure TThreadPool.RunThread(ThreadIndex : Integer);
var
  Obj : PThreadObj;
  TaskIndex,idx : Integer;
begin
  with _TASK_LIST.LockList do
  try
    if Count>0 then
    begin
      case _RUN_ORDER of
      roNormal : TaskIndex := Count-1;
      roRandom : begin
                   if Count>1 then
                     TaskIndex := Random(Count)
                   else
                     TaskIndex := 0;
                 end;
      else TaskIndex := Count-1;
      end;
      CreateThread(ThreadIndex,Items[TaskIndex]);
      _THREADS[ThreadIndex]._THREAD_INDEX := ThreadIndex;
      _THREADS[ThreadIndex]._TASK_INDEX := PTaskObj(Items[TaskIndex])^.Index;
      _THREADS[ThreadIndex]._THREAD_LIST := _THREAD_LIST;
      _THREADS[ThreadIndex].OnTerminate := TaskDone;
      SetThread(ThreadIndex,Items[TaskIndex]);
      idx := PTaskObj(Items[TaskIndex])^.Index;
      FreeMem(PTaskObj(Items[TaskIndex])^.Param);
      FreeMem(Items[TaskIndex]);
      Delete(TaskIndex);
      if Count = 0 then
      _TASK_OVER := True;
      with _THREAD_LIST.LockList do
      try
        GetMem(Obj,SizeOf(TThreadObj));
        Obj^.Handle := _THREADS[ThreadIndex].Handle;
        Obj^.Index := ThreadIndex;
        Obj^.Active := True;
        Obj^.TIndex := idx;
        Add(Obj);
      finally
        _THREAD_LIST.UnlockList;
      end;
      _THREADS[ThreadIndex].Resume;
    end;
  finally
    _TASK_LIST.UnlockList;
  end;
end;

procedure TThreadPool.CreateThread(ThreadIndex : Integer ; Task : Pointer);
begin
  _THREADS[ThreadIndex] := _THREAD_CLASS.Create;
end;

procedure TThreadPool.CreateThreads;
var
  i,TaskCount : Integer;
begin
  with _TASK_LIST.LockList do
  try
    TaskCount := Count;
    if TaskCount < _THREAD_COUNT then
    _THREAD_COUNT := TaskCount;
  finally
    _TASK_LIST.UnlockList;
  end;
  //
  if _TERM_PRO_BAR <> nil then
  InitProgressBar(_TERM_PRO_BAR,_THREAD_COUNT);
  //
  if _PROGRESS_BAR <> nil then
  InitProgressBar(_PROGRESS_BAR,TaskCount);
  //
  SetLength(_THREADS,_THREAD_COUNT);
  for i:=Low(_THREADS) to High(_THREADS) do
  RunThread(i);
end;

procedure TThreadPool.Start;
begin
  if _THREAD_COUNT>0 then
  CreateThreads;
  TraceLog('线程已经全部启动！');
end;

procedure TThreadPool.Pause;
var
  i : Integer;
begin
  for i:= Low(_THREADS) to High(_THREADS) do
    if _THREADS[i]<>nil then
    _THREADS[i].Pause;
end;

procedure TThreadPool.Goon;
var
  i : Integer;
begin
  for i:= Low(_THREADS) to High(_THREADS) do
    if _THREADS[i]<>nil then
    _THREADS[i].Goon;
end;

procedure TThreadPool.Stop;
var
  i : Integer;
begin
  _STOP_FLAG := True;                          // Modified 2006-7-18 18:49:46
  for i:= Low(_THREADS) to High(_THREADS) do
  if _THREADS[i]<>nil then
  begin
//    TraceLog(Format('Terminating Thread %d',[i]));
    try
      _THREADS[i].Terminate;
      _THREADS[i].Stop;
    except
    end;
  end;
end;

procedure LoadTask(Index : Integer; Param : Pointer; TaskList : TThreadList);
var
  TaskObj : PTaskObj;
begin
  GetMem(TaskObj,SizeOf(TTaskObj));
  TaskObj^.Index := Index;
  TaskObj^.Param := Param;
  TaskList.Add(TaskObj);
end;

{THTTPtaskPool}
procedure THTTPtaskPool.SetThread(ThreadIndex : Integer; Task : Pointer);
begin
  inherited;
  with THTTPtask(_THREADS[ThreadIndex]) do
  begin
    _READ_TIME_OUT := P_READ_TIME_OUT;
    _CONN_TIME_OUT := P_CONN_TIME_OUT;
    _OnTraceLog := P_HTTPtaskTrace;
    _USE_PROXY := P_USE_PROXY;
    if _USE_PROXY then
    begin
      _PROXY_ADDR := P_PROXY_ADDR;
      _PROXY_PORT := P_PROXY_PORT;
      _NEED_AUTH := P_NEED_AUTH;
      if _NEED_AUTH then
      begin
        _PROXY_USER := P_PROXY_USER;
        _PROXY_PASS := P_PROXY_PASS;
      end;
    end;
  end;
end;

end.