Delphi: многопоточный список потоковых заданий - организация очередей

Question

У меня есть некоторые операции, основанные на TThreads. Теперь мне нужно создать поток, содержащий список заданий, которые нужно выполнить, а затем запустить каждое из них, как только закончится предыдущий ... Как мне его написать? Я не могу допустить одновременного запуска потоков, поскольку может быть выполнено более 10 000 операций. Довольно сложно найти документированные примеры TEvent и других синхронизирующих объектов ... Надеюсь, я найду здесь помощь ...

Спасибо заранее, Михал

Answer 1

Не основывайте свои операции на потоках. Это неправильный дизайн. Вместо этого вы должны создать базовый класс для вашей операции, который предоставляет метод для выполнения операции. Напишите классы-потомки для реализации конкретных операций. Не делайте никаких предположений о контексте потоков, всегда используйте критические секции или подобные объекты синхронизации для защиты доступа к общим ресурсам. Что еще более важно, старайтесь избегать общих ресурсов или, по крайней мере, сделайте общие ресурсы доступными только для чтения, чтобы блокировка не требовалась.

С этим дизайном становится возможным выполнять каждую операцию в потоке VCL, вызывая метод операции напрямую, чтобы использовать класс-потомок TThread для выполнения операции в своем собственном потоке (что у вас, кажется, сейчас) или запланировать все операции в пуле потоков. Количество потоков в пуле может быть отрегулировано во время выполнения, чтобы соответствовать характеру операций (с привязкой к процессору или с вводом / выводом) и количеством ядер процессора, имеющихся в системе. И чтобы ответить на ваш вопрос: даже можно полностью сериализовать операции, вынудив пул использовать один поток. По сути, вы можете полностью изменить способ выполнения ваших операций, не меняя их .

Answer 2

Вы должны посмотреть на OmniThreadLibrary . Это делает рабочие потоки чрезвычайно простыми, и вы просто добавляете для этого задачу; каждый начинается сразу после завершения предыдущего, а библиотека позволяет легко передавать сообщения о состоянии и т. д. обратно в основной поток для обновлений пользовательского интерфейса.

Answer 3

Я только что реализовал нечто очень похожее.

Я думаю, что вам нужна система пула потоков.

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

Преимущества этого подхода состоят в том, что вы можете легко контролировать максимальное количество одновременных операций и не создавать и не отбрасывать потоки для каждой операции (что дорого).

Answer 4

Допустим, у вас есть поток, который выполнит остальные, скажем WorkerThread.

В WorkerThread вы можете поместить вызываемые потоки в массив TThread, TThreadList, TList, в основном, в зависимости от того, что вам удобнее.

Затем в цикле for запустите каждый из них. Теперь, так как вы не хотите, чтобы они запускались одновременно, у вас есть 2 способа ожидания запуска потока, либо используйте какой-либо флаг и слушайте событие OnTerminate потока, которое при получении запуска устанавливает флаг, или используйте

WaitForSingleObject(Thread.Handle, INFINITE);

Answer 5

Generics Collections TQueue может использоваться в качестве контейнера для отдельных объектов задания.

Рабочий поток затем выбирает ('извлекает') первое задание и выполняет его, а затем продолжает работу до тех пор, пока очередь не станет пустой - тогда ему придется приостановить работу, пока другое задание не будет добавлено в очередь, и продолжить.

Для добавления новых заданий вам потребуется реализовать потокобезопасный доступ к очереди.

Answer 6

Я не до конца понимаю ваш вопрос и на самом деле не понимаю смысла в использовании нескольких потоков и затем сериализации выполнения ... но вы можете сделать это так:

 JobList : TList <TThread>;
 ...
 JobList.Add (TMyCustomThread.Create (True));  // Create suspended
 JobList.Add (TMyOtherThread.Create (True));   // Create suspended
 ...
 for Thread in JobList do
   begin
   Thread.Start;
   Thread.WaitFor;
   end;

Это выполнит каждый поток, а затем дождется окончания потока перед выполнением следующего потока.

В примере кода предполагается, что вы используете D2009 или более позднюю версию (вы не указали в своем вопросе) и что вы создали свои потоки в приостановленном состоянии. Если вы используете более старую версию Delphi, вам нужно вызвать Resume вместо Start и заменить общий TList на простой TList или массив.

Помните об утечках памяти в примере кода.