Джулиан Бакнелл - Фундаментальные алгоритмы и структуры данных в Delphi

FSyncObj : TtdProduceConsumeSync;

protected

procedure Execute; override;

public

constructor Create(aStream : TStream;

aSyncObj : TtdProduceConsumeSync;

aBuffers : TQueuedBuffers);

end;

constructor TProducer.Create(aStream : TStream;

aSyncObj : TtdProduceConsumeSync;

aBuffers : TQueuedBuffers);

begin

inherited Create (true);

FStream := aStream;

FSyncObj :=,aSyncObj;

FBuffers aBuffers;

end;

procedure TProducer.Execute;

var

Tail : PBuffer;

begin

{выполнять до момента опустошения потока...}

repeat

{сигнализировать о готовности к началу генерирования данных}

FSyncObj.StartProducing;

{считать блок из потока в конечный буфер}

Tail FBuffers.Tail;

Tail^.bCount := FStream.Read(Tail^.bBlock, BufferSize);

{переместить указатель конца очереди}

FBuffers.AdvanceTail;

{поскольку выполняется запись нового буфера, необходимо сигнализировать о созданных данных}

FSyncObj.StopProducing;

until (Tail^.bCount ? 0);

end;

type

TConsumer = class(TThread) private

FBuffers : TQueuedBuffers;

FStream : TStream;

FSyncObj : TtdProduceConsumeSync;

protected

procedure Execute; override;

public

constructor Create(aStream : TStream;

aSyncObj : TtdProduceConsumeSync;

aBuffers : TQueuedBuffers);

end;

constructor TConsumer.Create(aStream : TStream;

aSyncObj : TtdProduceConsumeSync;

aBuffers : TQueuedBuffers);

begin

inherited Create (true);

FStream := aStream;

FSyncObj := aSyncObj;

FBuffers := aBuffers;

end;

procedure TConsumer.Execute;

var

Head : PBuffer;

begin

{сигнализировать о готовности к началу потребления данных}

FSyncObj.StartConsuming;

{извлечь начальный буфер}

Head := FBuffers.Head;

{до тех пор, пока начальный буфер не опустошен...}

while (Head^.bCount <> 0) do

begin

{выполнить запись блока из начального буфера в поток}

FStream.Write(Head^.bBlock, Head^.bCount);

{переместить указатель начала очереди}

FBuffers.AdvanceHead;

{поскольку было выполнено считывание и обработка буфера, необходимо сообщить о том, что данные были использованы}

FSyncObj.StopConsuming;

{сигнализировать о готовности снова приступить к потреблению данных}

FSyncObj.StartConsuming;

{извлечь начальный буфер}

Head := FBuffers.Head;

end;

end;

И, наконец, мы можем рассмотреть подпрограмму копирования потока, приведенную в листинге 12.14. Она принимает два параметра: входной поток и выходной поток. Подпрограмма создает специальный объект типа TQueuedBuffers. Этот объект содержит все ресурсы и методы, необходимые для реализации организованного в виде очереди набора буферов. Он создает также экземпляр класса TtdProducerConsumerSync, который будет действовать в качестве объекта синхронизации, обеспечивающего согласованную работу производителя и потребителя.

Листинг 12.14. Многопоточное копирование

procedure ThreadedCopyStream(aSrcStream, aDestStream : TStream);

var

SyncObj : TtdProduceConsumeSync;

Buffers : TQueuedBuffers;

Producer : TProducer;

Consumer : TConsumer;

WaitArray : array [ 0..1] of THandle;

begin

SyncObj := nil;

Buffers := nil;

Producer :=nil;

Consumer :=nil;

try

{создать объект синхронизации, объект организованных в виде очереди буферов (с 20 буферами) и два потока}

SyncObj := TtdProduceConsumeSync.Create(20);

Buffers := TQueuedBuffers.Create(20);

Producer := TProducer.Create(aSrcStream, SyncObj, Buffers);

Consumer := TConsumer.Create(aDestStream, SyncObj, Buffers);

{сохранить дескрипторы потоков, что обеспечивает возможность ожидания их передачи}

WaitArray[0] := Producer.Handle;

WaitArray[1] := Consumer.Handle;

{запустить потоки}

Consumer.Resume;

Producer.Resume;

{ожидать окончания потоков}

WaitForMultipleObjects(2, @WaitArray, true, INFINITE);

finally

Producer.Free;

Consumer.Free;

Buffers.Free;

SyncObj.Free;

end;

end;

Затем подпрограмма копирования создает два потока, между которыми будет выполняться копирование, и возобновляет их выполнение (потоки создаются в приостановленном состоянии). Далее подпрограмма дожидается завершения обоих потоков и выполняет очистку. Полный код подпрограммы можно найти в файлах TstCopy.dpr и TstCopyu.pas на web-сайте издательства, в разделе материалов.

Реализовать рассмотренное приложение, в котором используется модель "производитель-потребитель", было достаточно просто. Теперь рассмотрим модель с одним производителем и несколькими потребителями. В этом случае имеется поток, который создает данные. Предположим, что существует несколько потоков, которым требуется считывать созданные данные. В упомянутом ранее примере использовались два потребителя, которые сжимали данные с применением разных алгоритмов. Еще одним примером мог бы служить браузер. Будем считать, что производитель выгружает web-страницу из удаленного сайта, а один потребитель считывает HTML-код, чтобы выполнить его сохранение на диске, второй считывает код для его отображения на экране, а третий - с целью отображения индикатора выполнения. Создание этих процессов как отдельных потребителей упрощает написание кода, поскольку каждый процесс должен выполнять только одну задачу.