Как предотвратить дублирование значений в общей очереди

Question

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

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

[править] Кто-то задал хороший вопрос по IRC, и я решил добавить ответ здесь.Вопрос был: «Почему ваши производители производят дубликаты?»Ответ в основном заключается в том, что производитель создает дубликаты, потому что мы не отслеживаем состояние «отправки» каждого объекта, только «отправлено» или «неотправлено».

Есть ли способ проверить наличие дубликатов вочередь?

Answer 1

Мне кажется, что на самом деле не проблема иметь дубликаты объектов в очереди; Вы просто хотите убедиться, что выполняете обработку только один раз для каждого объекта.

Если это так, я предлагаю, чтобы потребительские потоки использовали набор для отслеживания обнаруженных объектов. Если объект отсутствует в наборе, добавьте его и обработайте; если он есть в наборе, игнорируйте его как дубликат.

Если это будет долговременная система, а не набор, используйте OrderedDict для отслеживания видимых объектов. Затем время от времени очищайте самые старые записи в OrderedDict.

Answer 2

Если говорить о классах в модуле Queue: следуя API, невозможно определить, содержит ли очередь данный объект.

Answer 3

Что вы подразумеваете под пометкой объекта как завершенного? Вы оставляете объект в очереди и меняете флаг? Или вы имеете в виду, что вы помечаете объект как завершенный в хранилище данных. Если первое, как очередь становится пустой? Если последнее, почему бы не удалить объект из очереди перед началом обработки?

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