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

SyncObj.Free;

end;

end;

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

Модель с одним производителем и несколькими потребителями

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

Итак, что же требуется, чтобы объект синхронизации поддерживал согласованную работу производителя и потребителей? Во-первых, производитель должен сообщать всем потребителям о наличии данных для считывания. Предположительно скорости работы потребителей будут различными, и поэтому они будут обрабатывать данные с различной частотой. Это предполагает существование по одному семафору "имеются данные" на каждый потребитель. Будем считать, что существует список буферов, которые производитель должен пополнять данными. И более того, этот список организован в виде циклической очереди. Следовательно, нам нужен единственный указатель конца очереди (управляемый исключительно производителем) и по одному указателю начала очереди для каждого потребителя, поскольку, по всей вероятности, каждый потребитель будет считывать буфера с различной частотой.

Так как же быть с производителем? Каким образом он узнает, что можно снова заполнять буфер данных? Понятно, что он может это делать только после того, как последний (предположительно самый медленный) потребитель прочитал достаточный объем данных, чтобы появилось место для его заполнения новыми данными (иначе говоря, как только буфер снова освободится). Это, в свою очередь, предполагает, что должен существовать счетчик потребителей для каждого буфера данных. Каждый раз, когда потребитель считывает данные из буфера, он уменьшает значение этого счетчика (число потребителей, которым еще только предстоит выполнить считывание данных из этого буфера). Таким образом, когда последний потребитель приступает к использованию каких-либо данных, известно, что он является последним, поскольку после уменьшения значение счетчика должно быть равно нулю. Обратите внимание, что потребители являются потоками и, следовательно, уменьшение значения счетчика следует выполнять безопасным для потоков образом.

Код этого расширенного класса TtdProduceManyConsumeSync, который позволяет нескольким потребителям потреблять данные, сгенерированные единственным производителем, приведен в листинге 12.15. Предполагается, что каждый поток потребителя имеет уникальный, начинающийся с нуля, идентификатор (на практике этого легко добиться, но при необходимости класс можно было бы расширить, чтобы потребители могли регистрироваться и отменять свою регистрацию, и чтобы идентификаторы присваивались им "на лету"). Затем потребитель использует этот идентификатор (числовое значение) при обращении к методам StartConsumer и StopConsumer.

Листинг 12.15. Класс синхронизации одного производителя и нескольких потребителей

В этом классе предполагается, что производитель заполняет буфера, которые затем используются потребителями. Буфера не имеют никакой реальной реализации в самом классе. Их предоставление - задача пользователя класса.

Метод StartProducing, показанный в листинге 12.16, работает во многом аналогично описанному в предыдущем случае: он просто дожидается передачи ему семафора "требуются данные". (Этот семафор содержит значение, равное количеству буферов, что позволяет производителю заполнить все буфера.)

type

TtdProduceManyConsumeSync = class private

FBufferCount : integer;

{счетчик буферов данных}

FBufferInfo : TList;

{циклическая очередь информации о буферах}

FBufferTail : integer;

{конец циклической очереди буферов}

FConsumerCount : integer;

{счетчик потребителей}

FConsumerInfo : TList;

{информация для каждого потребителя}

FNeedsData : THandle;

{семафор}

protected

public

constructor Create(aBufferCount : integer;

aConsumerCount : integer);

destructor Destroy; override;

procedure StartConsuming(aid : integer);

procedure StartProducing;