TThreadedQueue не способен к нескольким потребителям?

Ну, трудно быть уверенным без большого тестирования, но, похоже, это ошибка, либо в TThreadedQueue, либо в TMonitor. Так или иначе, это RTL, а не ваш код. Вы должны указать это как отчет о контроле качества и использовать свой пример выше как код «как воспроизвести».

Я рекомендую вам использовать OmniThreadLibrary http://www.thedelphigeek.com/search/label/OmniThreadLibrary при работе с streamами, параллелизмом и т. Д. Primoz проделал очень хорошую работу, и на сайте вы найдете много полезной документации ,

Ваш пример, похоже, отлично работает под XE2, но если мы заполняем вашу очередь, он терпит неудачу с AV на PushItem. (Проверено под обновлением XE21)

Чтобы воспроизвести, просто увеличивайте создание своей задачи с 100 до 1100 (ваша глубина очереди была установлена ​​на 1024)

 for i:= 1 to 1100 do fThreadQueue.PushItem( TThreadTaskMsg.Create( i,'')); 

Это всегда умирает для меня каждый раз в Windows 7. Сначала я попытался настойчиво протестировать его, и он потерпел неудачу в цикле 30 … затем в цикле 16 … затем на 65, так что с разными интервалами, но он последовательно терпел неудачу при некоторых точка.

  iLoop := 0; while iLoop < 1000 do begin Inc(iLoop); WriteLn('Loop: ' + IntToStr(iLoop)); for i:= 1 to 100 do fThreadQueue.PushItem( TThreadTaskMsg.Create( i,'')); end; 

Я искал class TThreadedQueue, но, похоже, не имел его в своем D2009. Я не собираюсь убивать себя над этим – поддержка streamа Delphi всегда была ошибкой .. errm … «неоптимальным», и я подозреваю, что TThreadedQueue ничем не отличается 🙂

Зачем использовать дженерики для объектов ПК (Продюсер / Потребитель)? Простой потомок TObjectQueue будет преуспевать – использовать его на протяжении десятилетий – отлично работает с несколькими производителями / потребителями:

 unit MinimalSemaphorePCqueue; { Absolutely minimal PC queue based on TobjectQueue and a semaphore. The semaphore count reflects the queue count 'push' will always succeed unless memory runs out, then you're stuft anyway. 'pop' has a timeout parameter as well as the address of where any received object is to be put. 'pop' returns immediately with 'true' if there is an object on the queue available for it. 'pop' blocks the caller if the queue is empty and the timeout is not 0. 'pop' returns false if the timeout is exceeded before an object is available from the queue. 'pop' returns true if an object is available from the queue before the timeout is exceeded. If multiple threads have called 'pop' and are blocked because the queue is empty, a single 'push' will make only one of the waiting threads ready. Methods to push/pop from the queue A 'semaHandle' property that can be used in a 'waitForMultipleObjects' call. When the handle is signaled, the 'peek' method will retrieve the queued object. } interface uses Windows, Messages, SysUtils, Classes,syncObjs,contnrs; type pObject=^Tobject; TsemaphoreMailbox=class(TobjectQueue) private countSema:Thandle; protected access:TcriticalSection; public property semaHandle:Thandle read countSema; constructor create; virtual; procedure push(aObject:Tobject); virtual; function pop(pResObject:pObject;timeout:DWORD):boolean; virtual; function peek(pResObject:pObject):boolean; virtual; destructor destroy; override; end; implementation { TsemaphoreMailbox } constructor TsemaphoreMailbox.create; begin {$IFDEF D2009} inherited Create; {$ELSE} inherited create; {$ENDIF} access:=TcriticalSection.create; countSema:=createSemaphore(nil,0,maxInt,nil); end; destructor TsemaphoreMailbox.destroy; begin access.free; closeHandle(countSema); inherited; end; function TsemaphoreMailbox.pop(pResObject: pObject; timeout: DWORD): boolean; // dequeues an object, if one is available on the queue. If the queue is empty, // the caller is blocked until either an object is pushed on or the timeout // period expires begin // wait for a unit from the semaphore result:=(WAIT_OBJECT_0=waitForSingleObject(countSema,timeout)); if result then // if a unit was supplied before the timeout, begin access.acquire; try pResObject^:=inherited pop; // get an object from the queue finally access.release; end; end; end; procedure TsemaphoreMailbox.push(aObject: Tobject); // pushes an object onto the queue. If threads are waiting in a 'pop' call, // one of them is made ready. begin access.acquire; try inherited push(aObject); // shove the object onto the queue finally access.release; end; releaseSemaphore(countSema,1,nil); // release one unit to semaphore end; function TsemaphoreMailbox.peek(pResObject: pObject): boolean; begin access.acquire; try result:=(count>0); if result then pResObject^:=inherited pop; // get an object from the queue finally access.release; end; end; end. - unit MinimalSemaphorePCqueue; { Absolutely minimal PC queue based on TobjectQueue and a semaphore. The semaphore count reflects the queue count 'push' will always succeed unless memory runs out, then you're stuft anyway. 'pop' has a timeout parameter as well as the address of where any received object is to be put. 'pop' returns immediately with 'true' if there is an object on the queue available for it. 'pop' blocks the caller if the queue is empty and the timeout is not 0. 'pop' returns false if the timeout is exceeded before an object is available from the queue. 'pop' returns true if an object is available from the queue before the timeout is exceeded. If multiple threads have called 'pop' and are blocked because the queue is empty, a single 'push' will make only one of the waiting threads ready. Methods to push/pop from the queue A 'semaHandle' property that can be used in a 'waitForMultipleObjects' call. When the handle is signaled, the 'peek' method will retrieve the queued object. } interface uses Windows, Messages, SysUtils, Classes,syncObjs,contnrs; type pObject=^Tobject; TsemaphoreMailbox=class(TobjectQueue) private countSema:Thandle; protected access:TcriticalSection; public property semaHandle:Thandle read countSema; constructor create; virtual; procedure push(aObject:Tobject); virtual; function pop(pResObject:pObject;timeout:DWORD):boolean; virtual; function peek(pResObject:pObject):boolean; virtual; destructor destroy; override; end; implementation { TsemaphoreMailbox } constructor TsemaphoreMailbox.create; begin {$IFDEF D2009} inherited Create; {$ELSE} inherited create; {$ENDIF} access:=TcriticalSection.create; countSema:=createSemaphore(nil,0,maxInt,nil); end; destructor TsemaphoreMailbox.destroy; begin access.free; closeHandle(countSema); inherited; end; function TsemaphoreMailbox.pop(pResObject: pObject; timeout: DWORD): boolean; // dequeues an object, if one is available on the queue. If the queue is empty, // the caller is blocked until either an object is pushed on or the timeout // period expires begin // wait for a unit from the semaphore result:=(WAIT_OBJECT_0=waitForSingleObject(countSema,timeout)); if result then // if a unit was supplied before the timeout, begin access.acquire; try pResObject^:=inherited pop; // get an object from the queue finally access.release; end; end; end; procedure TsemaphoreMailbox.push(aObject: Tobject); // pushes an object onto the queue. If threads are waiting in a 'pop' call, // one of them is made ready. begin access.acquire; try inherited push(aObject); // shove the object onto the queue finally access.release; end; releaseSemaphore(countSema,1,nil); // release one unit to semaphore end; function TsemaphoreMailbox.peek(pResObject: pObject): boolean; begin access.acquire; try result:=(count>0); if result then pResObject^:=inherited pop; // get an object from the queue finally access.release; end; end; end. - unit MinimalSemaphorePCqueue; { Absolutely minimal PC queue based on TobjectQueue and a semaphore. The semaphore count reflects the queue count 'push' will always succeed unless memory runs out, then you're stuft anyway. 'pop' has a timeout parameter as well as the address of where any received object is to be put. 'pop' returns immediately with 'true' if there is an object on the queue available for it. 'pop' blocks the caller if the queue is empty and the timeout is not 0. 'pop' returns false if the timeout is exceeded before an object is available from the queue. 'pop' returns true if an object is available from the queue before the timeout is exceeded. If multiple threads have called 'pop' and are blocked because the queue is empty, a single 'push' will make only one of the waiting threads ready. Methods to push/pop from the queue A 'semaHandle' property that can be used in a 'waitForMultipleObjects' call. When the handle is signaled, the 'peek' method will retrieve the queued object. } interface uses Windows, Messages, SysUtils, Classes,syncObjs,contnrs; type pObject=^Tobject; TsemaphoreMailbox=class(TobjectQueue) private countSema:Thandle; protected access:TcriticalSection; public property semaHandle:Thandle read countSema; constructor create; virtual; procedure push(aObject:Tobject); virtual; function pop(pResObject:pObject;timeout:DWORD):boolean; virtual; function peek(pResObject:pObject):boolean; virtual; destructor destroy; override; end; implementation { TsemaphoreMailbox } constructor TsemaphoreMailbox.create; begin {$IFDEF D2009} inherited Create; {$ELSE} inherited create; {$ENDIF} access:=TcriticalSection.create; countSema:=createSemaphore(nil,0,maxInt,nil); end; destructor TsemaphoreMailbox.destroy; begin access.free; closeHandle(countSema); inherited; end; function TsemaphoreMailbox.pop(pResObject: pObject; timeout: DWORD): boolean; // dequeues an object, if one is available on the queue. If the queue is empty, // the caller is blocked until either an object is pushed on or the timeout // period expires begin // wait for a unit from the semaphore result:=(WAIT_OBJECT_0=waitForSingleObject(countSema,timeout)); if result then // if a unit was supplied before the timeout, begin access.acquire; try pResObject^:=inherited pop; // get an object from the queue finally access.release; end; end; end; procedure TsemaphoreMailbox.push(aObject: Tobject); // pushes an object onto the queue. If threads are waiting in a 'pop' call, // one of them is made ready. begin access.acquire; try inherited push(aObject); // shove the object onto the queue finally access.release; end; releaseSemaphore(countSema,1,nil); // release one unit to semaphore end; function TsemaphoreMailbox.peek(pResObject: pObject): boolean; begin access.acquire; try result:=(count>0); if result then pResObject^:=inherited pop; // get an object from the queue finally access.release; end; end; end. 

Я не думаю, что TThreadedQueue должен поддерживать нескольких потребителей. Это FIFO, в соответствии с файлом справки. У меня создается впечатление, что есть одна нить, а другая (только одна!) Появляется.