Как сделать параллельный итератор для какого-либо источника?

Question

Я хотел бы иметь итератор, который может считываться несколькими потоками одновременно, чтобы я мог обрабатывать данные источника итератора параллельно.Проблема в том, что я не могу соединить hasNext() с его логическим next(), поскольку они могут идти в разные потоки.(То есть два потока могут вызвать hasNext(), каждый из которых видит true, а затем второй поток завершается с ошибкой, потому что был только один элемент.) Моя проблема в том, что для некоторых источников я не знаю, есть ли у него следующий элементпока я не попытаюсь это прочитать.Одним из таких примеров является чтение строк из файла;другой читает Term экземпляров из индекса Lucene.

Я думал о настройке очереди внутри итератора и заполнении очереди отдельным потоком.Таким образом, hasNext() реализуется с точки зрения размера очереди.Но я не понимаю, как я могу гарантировать, что очередь будет заполнена, потому что этот поток может истощиться.

Должен ли я игнорировать контракт Итератора и просто вызывать next() до тех пор, пока не будет выброшено NoSuchElementException?

Есть ли более элегантный способ решения проблемы?

Answer 1

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

Передайте LinkedBlockingQueue, и ваши потоки будут выполнять queue.poll (), пока ничего не останется.

Answer 2

Мне приходит в голову один обходной путь, чтобы сохранить (большую часть) контракт и избежать NoSuchElementExceptions: iterator.next() может вернуть пользовательский объект-маркер "Конец итерации", который может быть обработан, но ничего, кроме манекена. Таким образом, если один поток получает true для hasNext(), но другой поток уже захватил последний элемент, то первый поток получит фиктивный (вместо исключения).

Вы должны быть в состоянии использовать этот тип итератора во всех обычных случаях использования, и однопоточные приложения должны даже заметить разницу. Должен использоваться и с расширенным циклом for.

Он потерпит неудачу, только если попытаться дождаться NoSuchElementException вместо проверки hasNext(), потому что это исключение не будет выброшено из-за фиктивных элементов.

Answer 3

Выбранный ответ будет работать, но он вводит сложность и потенциальную ненужную буферизацию. Почему бы не игнорировать Iterator контракт и написать свой собственный:

public interface ConcurrentIterator<T> {

    T next() throws EndOfIterationException;

}

Это будет поточно-ориентированным, если ваша реализация. Можно даже завернуть в него Iterator.

Answer 4

Я мог упустить момент, но разве вы не могли использовать синхронизированный блок в этой ситуации?

synchronized(iterator)
{
    if (iterator.hasNext()) element = iterator.next();
}

Здесь, когда один поток использует итератор, другие потоки не смогут получить доступэто.