Java 8 параллельный поток + anyMatch - ли потоки прерываются после совпадения?

Если у меня есть параллельный поток в java 8, и я заканчиваю с anyMatch, и моя коллекция имеет элемент, который соответствует предикату, я пытаюсь выяснить, что происходит, когда один поток обрабатывает этот элемент.

Я знаю, что anyMatch имеет короткое замыкание, поэтому я не ожидал, что последующие элементы будут обработаны после обработки соответствующего элемента. Моя путаница связана с тем, что происходит с другими потоками, которые предположительно находятся в середине элементов обработки. Я могу думать о 3 вероятных сценариях: а) Они прерваны? б) Продолжают ли они обрабатывать элемент, над которым они работают, а затем, когда все нити ничего не делают, я получаю свой результат? c) Получил ли я свой результат, но потоки, которые обрабатывали другие элементы, продолжали обрабатывать эти элементы (но не брать другие элементы после их завершения)?

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

Спасибо

Ответы

Ответ 1

После некоторого копания через исходный код Java, я думаю, что нашел ответ.

Другие потоки периодически проверяют, нашел ли другой поток ответ, и если да, то перестают работать и отменить еще не запущенные узлы.

java.util.Stream.FindOps$FindTask имеет этот метод:

private void foundResult(O answer) {
        if (isLeftmostNode())
            shortCircuit(answer);
        else
            cancelLaterNodes();
    }

Его родительский класс AbstractShortcircuitTask реализует shortCircuit следующим образом:

 /**
 * Declares that a globally valid result has been found.  If another task has
 * not already found the answer, the result is installed in
 * {@code sharedResult}.  The {@code compute()} method will check
 * {@code sharedResult} before proceeding with computation, so this causes
 * the computation to terminate early.
 *
 * @param result the result found
 */
protected void shortCircuit(R result) {
    if (result != null)
        sharedResult.compareAndSet(null, result);
}

И фактический compute() метод, который выполняет работу, имеет эту важную строку:

 AtomicReference<R> sr = sharedResult;
    R result;
    while ((result = sr.get()) == null) {
        ...//does the actual fork stuff here
    }

где sharedResult обновляется методом shortCircuit(), поэтому вычисление увидит его при следующем проверке условия цикла while.

ИЗМЕНИТЬ Итак, вкратце:

  • Темы не прерываются
  • Вместо этого они будут периодически проверять, нашел ли кто-нибудь ответ и прекратит дальнейшую обработку, если ответ был найден.
  • После обнаружения ответа новые потоки не будут запущены.