Delphi.RobustServer. Работа с длинными операциями

Короткие запросы хороши тем, что мы можем получить ответ сразу, а вот как обрабатывать длинные запросы? На StackOverflow Remy Lebeau советует дать ответ о старте работы клиенту сразу, и саму работу выносить в отдельный поток. Я уже писал на эту тему. В данном посте, попробовал реализовать эту идею, в применении к моему RobustServer и вот что у меня получилось.

Для начала я написал класс TLongTaskThread

  TNotifyLongTask = procedure(aProgress: double = 0.00; aMsg: string = '') of object;
  TNotifyLongTaskException = procedure(aEClass: string; aMsg: string) of object;

  TLongTaskThread = class(TThread)
  private
    FProgress: double;
    FGuid: string;
    FProc: TProc;
    FOnStart: TNotifyLongTask;
    FOnProgress: TNotifyLongTask;
    FOnFinish: TNotifyLongTask;
    FOnException: TNotifyLongTaskException;
  protected
    procedure Execute(); override;
  published
    constructor Create(aProc: TProc);
    function IsTerminated: Boolean;
    property Progress: double read FProgress write FProgress;
    property GuiId: string read FGuid;
    property OnStart: TNotifyLongTask read FOnStart write FOnStart;
    property OnProgress: TNotifyLongTask read FOnProgress write FOnProgress;
    property OnFinish: TNotifyLongTask read FOnFinish write FOnFinish;
    property OnException: TNotifyLongTaskException read FOnException write FOnException;
  end;

{ TLongTaskThread }

constructor TLongTaskThread.Create(aProc: TProc);
var
  guid: TGuid;
begin
  inherited Create(true);
  FreeOnTerminate := true;
  FProc := aProc;
  FGuid := TGuid.NewGuid.ToString();
end;

procedure TLongTaskThread.Execute;
begin
  inherited;
  try
    if not Terminated then
      FProc();
  except
    on E: Exception do
      if Assigned(OnException) then
        OnException(E.ClassName, E.Message);
  end;
end;

function TLongTaskThread.IsTerminated: Boolean;
begin
  Result := Terminated;
end;

TNotifyLongTask = procedure(aProgress: double = 0.00; aMsg: string = '') of object;

TNotifyLongTaskException = procedure(aEClass: string; aMsg: string) of object;

TLongTaskThread = class(TThread)

private

FProgress: double;

FGuid: string;

FProc: TProc;

FOnStart: TNotifyLongTask;

FOnProgress: TNotifyLongTask;

FOnFinish: TNotifyLongTask;

FOnException: TNotifyLongTaskException;

protected

procedure Execute(); override;

published

constructor Create(aProc: TProc);

function IsTerminated: Boolean;

property Progress: double read FProgress write FProgress;

property GuiId: string read FGuid;

property OnStart: TNotifyLongTask read FOnStart write FOnStart;

property OnProgress: TNotifyLongTask read FOnProgress write FOnProgress;

property OnFinish: TNotifyLongTask read FOnFinish write FOnFinish;

property OnException: TNotifyLongTaskException read FOnException write FOnException;

end;

{ TLongTaskThread }

constructor TLongTaskThread.Create(aProc: TProc);

var

guid: TGuid;

begin

inherited Create(true);

FreeOnTerminate := true;

FProc := aProc;

FGuid := TGuid.NewGuid.ToString();

end;

procedure TLongTaskThread.Execute;

begin

inherited;

try

if not Terminated then

FProc();

except

on E: Exception do

if Assigned(OnException) then

OnException(E.ClassName, E.Message);

end;

function TLongTaskThread.IsTerminated: Boolean;

begin

Result := Terminated;

end;

Он позволил мне создавать потоки, в которых выполнялась бы долгая работа. У каждого экземпляра такого класса есть Guid и Progress, которые бы позволяли отслеживать прогресс выполнения работ.

В Main, я создал экземпляр TThreadList, в который добавляю потоки, после их создания

FLongTaskThreads := TSP<TThreadList>.Create();

1	FLongTaskThreads := TSP<TThreadList>.Create();

Также добавил несколько событий, OnStart, OnProgress, OnFinish, на которые можно подписаться. Например отправлять ответный запрос о выполнении, отправлять информацию о выполнении на почту, ну и так далее, что как говорится душе угодно.

Посмотрим пример применения данного класса

procedure TRPTests.LongTask;
var
  jo: ISuperObject;
  t: TLongTaskThread;
begin
  t := TLongTaskThread.Create(
    procedure()
    var
      i: integer;
    begin
      if Assigned(t) and Assigned(t.OnStart) then
        (t.OnStart(i, 'ThreadStarted'));

      for i := 0 to 99 do
      begin
        sleep(1000);
        t.Progress := i;

        if TLongTaskThread(t).IsTerminated then
          break;

        if Assigned(t) and Assigned(t.OnProgress) then
          (t.OnProgress(i));

        // raise Exception.Create('Test Error Message'); // test Exception
            end;

      if Assigned(t) and Assigned(t.OnFinish) and not TLongTaskThread(t).IsTerminated then
        (t.OnFinish(i, ' ThreadFinished')); // thread finished
    end);

  t.OnStart := OnStartLongTask;
  t.OnProgress := OnProgressLongTask;
  t.OnFinish := OnFinishLongTask;
  t.OnException := OnException;

  Main.LongTaskThreads.Add(t);
  t.Start();
  if Assigned(t.OnStart) then
    (t.OnStart); // thread started

  jo := SO();
  jo.S['threadGuid'] := t.GuiId;
  FResponses.OkWithJson(jo.AsJSon(false, false));
end;

procedure TRPTests.LongTask;

var

jo: ISuperObject;

t: TLongTaskThread;

begin

t := TLongTaskThread.Create(

procedure()

var

i: integer;

begin

if Assigned(t) and Assigned(t.OnStart) then

(t.OnStart(i, 'ThreadStarted'));

for i := 0 to 99 do

begin

sleep(1000);

t.Progress := i;

if TLongTaskThread(t).IsTerminated then

break;

if Assigned(t) and Assigned(t.OnProgress) then

(t.OnProgress(i));

// raise Exception.Create('Test Error Message'); // test Exception

end;

if Assigned(t) and Assigned(t.OnFinish) and not TLongTaskThread(t).IsTerminated then

(t.OnFinish(i, ' ThreadFinished')); // thread finished

end);

t.OnStart := OnStartLongTask;

t.OnProgress := OnProgressLongTask;

t.OnFinish := OnFinishLongTask;

t.OnException := OnException;

Main.LongTaskThreads.Add(t);

t.Start();

if Assigned(t.OnStart) then

(t.OnStart); // thread started

jo := SO();

jo.S['threadGuid'] := t.GuiId;

FResponses.OkWithJson(jo.AsJSon(false, false));

end;

Я создаю экземпляр данного класса, по аналогии с анонимным потоком. В теле потока у нас TProc, внутри которой выполняется простой цикл с засыпанием, имитирующим долгую работу потока )))

Далее, мы подписываем события, которые нам нужны, добавляем поток в TThreadList, который позволит нам безопасно работать с потоками, запускаем наш поток в работу, и наблюдаем за прогрессом.

В ответ, как видно, мы получили ответ, что работа началась и GUID.

События я подписал самым простым способом

procedure TRPTests.OnStartLongTask(aProgress: double; aMsg: string);
begin
  //Main.mAnswer.Lines.Add(aProgress.ToString() + aMsg);
end;

procedure TRPTests.OnStartLongTask(aProgress: double; aMsg: string);

begin

//Main.mAnswer.Lines.Add(aProgress.ToString() + aMsg);

end;

procedure TRPTests.OnProgressLongTask(aProgress: double; aMsg: string);
begin
  Main.mAnswer.Lines.Add(aProgress.ToString() + aMsg);
end;

procedure TRPTests.OnProgressLongTask(aProgress: double; aMsg: string);

begin

Main.mAnswer.Lines.Add(aProgress.ToString() + aMsg);

end;

procedure TRPTests.OnFinishLongTask(aProgress: double; aMsg: string);
begin
  Main.mAnswer.Lines.Add(aProgress.ToString() + aMsg);
end;

procedure TRPTests.OnFinishLongTask(aProgress: double; aMsg: string);

begin

Main.mAnswer.Lines.Add(aProgress.ToString() + aMsg);

end;

procedure TRPTests.OnException(aClass, aMsg: string);
begin
// Notify exception here...
  // free allocated memory and send any notification
  Main.mAnswer.Lines.Add(aClass + ' ' + aMsg);
end;

procedure TRPTests.OnException(aClass, aMsg: string);

begin

// Notify exception here...

// free allocated memory and send any notification

Main.mAnswer.Lines.Add(aClass + ' ' + aMsg);

end;

По хорошему, здесь надо ещё сделать TThread.Synchronize, но это уже несколько другая задача и в сервер вряд ли кто-то будет отправлять события, рабочий вариант ответным запросом, на почту или ещё куда-то. Поэтому здесь, я этим заморачиваться не стал.

Получение прогресса запросом

Чтобы получить прогресс долгого запроса ещё одним запросом, я оформил такой метод

procedure TRPTests.GetLongTaskProgress(aGuid: string);
var
  p: double;
  json: ISuperObject;
  i: integer;
begin
  p := 0.00;
  with Main.LongTaskThreads.LockList() do
  try
    for i := 0 to Count - 1 do
      if TLongTaskThread(Items[i]).GuiId = aGuid then
      begin
        p := TLongTaskThread(Items[i]).Progress;
        break;
      end;
  finally
    Main.LongTaskThreads.UnlockList();
  end;
  json := SO;
  json.D['progress'] := p;
  FResponses.OkWithJson(json.AsJSon(false, false));
end;

procedure TRPTests.GetLongTaskProgress(aGuid: string);

var

p: double;

json: ISuperObject;

i: integer;

begin

p := 0.00;

with Main.LongTaskThreads.LockList() do

try

for i := 0 to Count - 1 do

if TLongTaskThread(Items[i]).GuiId = aGuid then

begin

p := TLongTaskThread(Items[i]).Progress;

break;

end;

finally

Main.LongTaskThreads.UnlockList();

end;

json := SO;

json.D['progress'] := p;

FResponses.OkWithJson(json.AsJSon(false, false));

end;

И вот как он у меня работает

То есть, в ответ json, внутри которого содержится прогресс нашего долгого запроса.

Обработка ошибок долгого запроса

Если в процессе работы долгого запроса произойдет ошибка, что делать в таком случае? Как было видно из кода выше, у нас есть специальное событие OnException, и всё, что нам нужно сделать – подписаться на это событие. В моем самом простом случае это выглядит так…

procedure TRPTests.OnException(aClass, aMsg: string);
begin
// Notify exception here...
  // free allocated memory and send any notification
  Main.mAnswer.Lines.Add(aClass + ' ' + aMsg);
end;

procedure TRPTests.OnException(aClass, aMsg: string);

begin

// Notify exception here...

// free allocated memory and send any notification

Main.mAnswer.Lines.Add(aClass + ' ' + aMsg);

end;

Посмотрим как это работает на практике. Раскомментирую часть кода, вызывающую ошибку

 raise Exception.Create('Test Error Message'); // test Exception

1	raise Exception.Create('Test Error Message'); // test Exception

И посмотрим на GUI нашего сервера

Ошибка пришла по назначению.

Закрытие сервера и уничтожение долгих потоков

Я сделал это таким образом. В части закрытия долгих потоков, закрытие сервера выглядит так

var
  i: integer;
begin
  with LongTaskThreads.LockList() do
  try
    for i := 0 to Count-1 do
    begin
      TLongTaskThread(Items[i]).FreeOnTerminate := false; // in other case they will be destroyed automatically
      TLongTaskThread(Items[i]).Free();
    end;
  finally
    LongTaskThreads.UnlockList();
  end;
end;

var

i: integer;

begin

with LongTaskThreads.LockList() do

try

for i := 0 to Count-1 do

begin

TLongTaskThread(Items[i]).FreeOnTerminate := false; // in other case they will be destroyed automatically

TLongTaskThread(Items[i]).Free();

end;

finally

LongTaskThreads.UnlockList();

end;

То есть, в нормальном случае, у нас поток бы отработал и уничтожился самостоятельно, потому что в конструкторе моего класса стоит

FreeOnTerminate := true

1	FreeOnTerminate := true

Но остановка / закрытие сервера это некий экстренный случай, и поэтому мы инвертируем эту строчку и затем, уничтожаем этот поток. Надо сказать, что метод Free у TThread перекрыт довольно грамотно, и там уже есть вся необходимая логика, которая делает Terminate потоку, и лишь затем его уничтожает

destructor TThread.Destroy;
begin
  if (FThreadID <> 0) and not FFinished and not FExternalThread then
  begin
    Terminate;
    if FCreateSuspended or FSuspended then
      Resume;
{$IFDEF MSWINDOWS}
    while not FStarted do
{$ELSE}
    while not ((not FCreateSuspended or FInitialSuspendDone) and FStarted) do
{$ENDIF}
      Yield;
    WaitFor;
  end;
  RemoveQueuedEvents(Self);
{$IF Defined(MSWINDOWS)}
  if (FHandle <> 0) and not FExternalThread then CloseHandle(FHandle);
{$ELSEIF Defined(POSIX)}
  // This final check is to ensure that even if the thread was never waited on
  // its resources will be freed.
  if (FThreadID <> 0) and not FExternalThread then pthread_detach(pthread_t(FThreadID));
{$IF Defined(LINUX)}
  sem_destroy(FCreateSuspendedSem);
{$ELSE}
  pthread_mutex_destroy(FCreateSuspendedMutex);
{$ENDIF LINUX or POSIX}
{$ENDIF POSIX}
  inherited Destroy;
  FFatalException.Free;
end;

destructor TThread.Destroy;

begin

if (FThreadID <> 0) and not FFinished and not FExternalThread then

begin

Terminate;

if FCreateSuspended or FSuspended then

Resume;

{$IFDEF MSWINDOWS}

while not FStarted do

{$ELSE}

while not ((not FCreateSuspended or FInitialSuspendDone) and FStarted) do

{$ENDIF}

Yield;

WaitFor;

end;

RemoveQueuedEvents(Self);

{$IF Defined(MSWINDOWS)}

if (FHandle <> 0) and not FExternalThread then CloseHandle(FHandle);

{$ELSEIF Defined(POSIX)}

// This final check is to ensure that even if the thread was never waited on

// its resources will be freed.

if (FThreadID <> 0) and not FExternalThread then pthread_detach(pthread_t(FThreadID));

{$IF Defined(LINUX)}

sem_destroy(FCreateSuspendedSem);

{$ELSE}

pthread_mutex_destroy(FCreateSuspendedMutex);

{$ENDIF LINUX or POSIX}

{$ENDIF POSIX}

inherited Destroy;

FFatalException.Free;

end;

Delphi.RobustServer. Работа с длинными операциями

Получение прогресса запросом

Обработка ошибок долгого запроса

Закрытие сервера и уничтожение долгих потоков

MY PROJECTS

Recent Posts

Categories