API потока Java 8: Исключения при изменении списков

Возьмем ArrayList и заполните его чем-то простым:

List<String> list = new ArrayList<>();
for (int i = 0; i < 10; i++) {
    list.add(""+i);
}

Я попытаюсь удалить один член, например, названный 5, с различными способами потока API. Для этого я определяю метод, который даст мне ConcurentModificationException при использовании традиционной итерации с итератором.

void removeMember(String clientListener) {
    list.remove(clientListener);
}

Этот код дает мне это исключение, которое я понимаю:

list.parallelStream()
    .filter(string -> string.equalsIgnoreCase("5"))
    .forEach(string -> removeMember(string));

Однако попытка просто stream(), а не parallelStream() дает исключение нулевого указателя (NPE), что для меня странно:

list.stream()
    .filter(string -> string.equalsIgnoreCase("5"))
    .forEach(string -> removeMember(string));

Теперь измените тип List на LinkedList<>. Последний код с stream() дает мне ConcurentModificationException, а parallelStream() внезапно срабатывает!

Итак, вопросы.

Является ли внутренняя кухня parallelStream() (разделители и другая магия) достаточно умной, чтобы использовать удаление этого элемента для LinkedList? Будет ли он работать всегда?
Почему этот NPE для ArrayList? Почему NPE, а не ConcurentModificationException Я имею в виду.

Ответ 1

Поведение вашего кода по существу undefined (следовательно, различные ответы, которые вы получаете). потоковая документация (раздел Non-Intereference) гласит:

Если источник потока не является параллельным, изменение источника данных потока во время выполнения конвейера потока может вызвать исключения, неправильные ответы или несоответствие поведения.

И ArrayList и LinkedList не являются параллельными.

Вы можете использовать параллельный источник, но было бы более целесообразно отказаться от изменения источника потока, например, используя Collection#removeIf:

list.removeIf(string -> string.equalsIgnoreCase("5"));

Ответ 2

Добавление некоторых отладочных отпечатков в конвейер показывает источник исключения NullPointerException:

list.stream().peek(string -> System.out.println("peek1 " + string)).filter(string -> string.equalsIgnoreCase("5")).peek(string -> System.out.println("peek2 " + string)).forEach(string -> removeMember(string));

Выводится:

peek1 0
peek1 1
peek1 2
peek1 3
peek1 4
peek1 5
peek2 5
peek1 7
peek1 8
peek1 9
peek1 null
Exception in thread "main" java.lang.NullPointerException
    at HelloWorld.lambda$main$1(HelloWorld.java:22)
    at HelloWorld$$Lambda$2/303563356.test(Unknown Source)
    at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:174)
    at java.util.stream.ReferencePipeline$11$1.accept(ReferencePipeline.java:373)
    at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1374)
    at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:512)
    at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:502)
    at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151)
    at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174)
    at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
    at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418)
    at HelloWorld.main(HelloWorld.java:22)

Когда "5" было удалено из Списка, все элементы от "6" до "9" были сдвинуты на одну позицию влево (т.е. их индексы были уменьшены на 1). Конвейер Stream не обнаружил его, поэтому он пропустил "6", и когда он обработал последнюю позицию (первоначально содержащую "9" ), она столкнулась с нулем, что привело к NullPointerException, когда для нее была оценена string.equalsIgnoreCase("5").

Это похоже на то, что вы получите в этом традиционном цикле for:

int size = list.size();
for (int i = 0; i < size; i++) {
    String string = list.get(i);
    if (string.equalsIgnoreCase("5"))
        removeMember(string);
}

Только здесь вы получите IndexOutOfBoundsException вместо NullPointerException, так как list.get(i) завершится с ошибкой i==9. Я думаю, что потоковый конвейер работает непосредственно во внутреннем массиве ArrayList, поэтому он не обнаруживает, что размер списка изменился.

EDIT:

Следуя комментарию Хольгера, я изменил код, чтобы устранить NullPointerException (изменив фильтр на filter(string -> "5".equalsIgnoreCase(string))). Это действительно дает ConcurrentModificationException:

peek1 0
peek1 1
peek1 2
peek1 3
peek1 4
peek1 5
peek2 5
peek1 7
peek1 8
peek1 9
peek1 null
Exception in thread "main" java.util.ConcurrentModificationException
    at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1380)
    at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:512)
    at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:502)
    at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151)
    at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174)
    at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
    at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418)
    at HelloWorld.main(HelloWorld.java:22)

Ответ 3

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

list.stream().filter(string -> !string.equalsIgnoreCase("5"))
                    .collect(Collectors.toList());

Что касается вашего другого вопроса о parallelStream и может ли этот подход работать всегда?

Нет, это определенно не будет. Используемые вами Lists не созданы для поддержки параллельного доступа, иногда он будет работать, в других случаях он будет терпеть неудачу, как вы видели, или дать вам "неожиданные" результаты. Если вы знаете, что к структуре данных будет обращен многократный поток, всегда кодируйте соответствующий код.

Ответ 4

При использовании Java8 Lambdas лучше не тратить трассировку стека на его номинальное значение. Ошибка должна быть прочитана, чтобы понять, что NPE вызван некоторой строкой кода в forEach lambda.So, вам нужно оценить каждую строку и посмотреть, что может быть причиной этого.