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

Question

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

Работа с потоками потребителей идет медленно.Кроме того, потребитель берет элемент из очереди посредством операции просмотра, и пока операция потребления не будет завершена, элемент не может быть удален из очереди.Это связано с тем, что поток производителя в качестве побочной операции также делает снимок всех элементов, которые не были полностью обработаны в тот момент времени.

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

Итак, опция использования стандартного ConcurrentLinkedQueue (который я используюпрямо сейчас) нет.

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

Какая структура данных больше всего подходит для этой проблемы?

Answer 1

Похоже, у вас должно быть две очереди:

• Необработанные
• В процессе

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

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

Answer 2

Я согласен с Джоном Скитом (+1) в том, что вам нужно два магазина для записи ожидающих и незавершенных предметов.Я бы использовал LinkedBlockingQueue, и каждый из ваших потребителей назвал бы take().Когда элемент поступает в очередь, его принимает один из потребителей.

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

Ваш производитель может сканировать HashSet, чтобы определить, что является выдающимся.