Stream API는 비상태 작업(필터, 맵)과 상태 작업(정렬, 고유함수, 제한)을 전체 입력을 처리해야 하는지의 여부에 따라 구분합니다. 병렬로 실행할 때 프레임워크는 소스 데이터를 여러 스레드로 분할하고 각 스레드는 개별적으로 세그먼트를 처리합니다. 소스 Spliterator가 ORDERED 특성을 보고하면 프레임워크는 만남 순서(소스에서 요소가 나타나는 순서)가 중요하다고 가정하고 파이프라인 전반에 걸쳐 이를 유지해야 합니다.
그러나 고유함수와 같은 상태 작업은 중복을 필터링하기 위해 전역 상태(본 이미 본 요소의 Set)에 의존합니다. 명시적인 만남 순서 집행이 없으면 병렬 스레드가 요소를 "첫 번째" 발생으로 주장하기 위해 경합할 수 있으며, 이로 인해 어떤 중복이 생존할지는 임의로 선택됩니다. 유사하게, 정렬은 전역 정렬을 요구하지만, 스트림이 무정렬로 표시되거나 소스가 ORDERED 특성을 결여한 경우 병렬 스레드의 중간 결과가 위치 보존 없이 병합될 수 있습니다. 이는 동일한 요소의 상대적 순서가 다르게 되거나 경우에 따라 출력 순서의 명백한 비결정론성을 유발할 수 있습니다.
해결책은 Spliterator 계약을 존중하는 것입니다: 만남 순서가 중요하면 소스는 ORDERED를 선언해야 하며, 파이프라인은 상태 작업 전에 **unordered()**를 호출해서는 안 됩니다. 고유함수의 경우, 이는 만남 순서에서 "첫 번째" 발생이 결정적으로 선택되도록 보장하며, 이로 인해 해당 단계에서 병렬성을 줄입니다. 순서가 중요하지 않다면 명시적으로 **unordered()**를 호출하여 프레임워크가 임의의 중복을 선택하고 동기화 없이 부분 결과를 병합하도록 허용함으로써 성능을 개선할 수 있으나, 이는 결정론을 희생하게 됩니다.
텔레메트리 처리 시스템은 각자의 나노초 타임스탬프와 고유한 센서 ID로 태그된 수백만 건의 센서 이벤트를 수집했습니다. 요구 사항은 센서 ID별로 이벤트를 중복 제거하면서 각 ID에 대해 연대기적으로 첫 이벤트를 보존하고 나머지는 타임스탬프에 따라 정렬하는 것이었습니다. 초기 구현은 **sensorReadings.parallelStream().distinct().sorted()**를 사용하여 ArrayList 소스가 삽입 순서를 유지하고 고유함수가 자연스럽게 첫 발생을 보존한다고 가정했습니다.
문제는 특정 센서 ID에 대한 "첫 번째" 이벤트가 다중 코어 하드웨어에서 실행될 때 원래 목록의 두 번째 또는 세 번째 발생으로 임의로 선택되는 간헐적인 테스트 실패로 나타났습니다. 조사를 통해 이 문제는 고유함수가 만남 순서 집행 없이 병렬로 실행되면서 발생했음을 결론지었습니다. 각 스레드는 목록의 일부를 처리하고 각 ID에 대한 자신의 지역 "첫 번째" 만남을 유지했습니다. 프레임워크가 이러한 부분 결과를 병합할 때 스레드의 전역 순서는 보장되지 않아 스레드 지역의 첫 번째 항목 중 임의의 선택이 발생했습니다.
세 가지 해결책이 평가되었습니다. 첫 번째 접근 방식은 전체 병렬성을 포기하고 순차 스트림으로 되돌아갔습니다. 이는 결정론적 동작을 복원하여 목록에서 가장 빠른 이벤트가 항상 이기도록 했습니다. 그러나 이는 최대 부하 시 처리 대기 시간을 400% 증가시켜 스루풋 SLA를 위반하고 예산이 잡히지 않은 하드웨어 업그레이드를 필요로 했습니다.
두 번째 접근 방식은 고유함수 전에 **.unordered()**를 삽입하여 중복을 허용한다고 명시적으로 신호를 보냈습니다. 이렇게 하면 스레드가 조정 없이 임의의 중복을 폐기할 수 있어 스루풋이 극대화되었습니다. 그러나 이는 가장 이른 판독값을 보존해야 하는 비즈니스 요구 사항을 위반하기 때문에 감사 기록에 수용할 수 없게 되었습니다.
세 번째 접근 방식은 **Collectors.toCollection(LinkedHashSet::new)**를 통해 스트림을 수집하여 하류 수집기로 LinkedHashSet을 활용했습니다. 이 방식은 스트림을 순서 있는 집합으로 변환하면서도 이전 필터 작업을 위한 병렬 분해를 허용했습니다. 그러나 이는 중간 고유함수 작업을 포기해야 했으며, 중복 제거 전에 전체 작업 집합을 유지하는 데 상당한 메모리를 소비했습니다.
선택된 솔루션은 파이프라인을 정렬된 단계와 무정렬 단계로 분리하는 것이었습니다. 시스템은 먼저 상태 없는 필터링 및 매핑 작업을 병렬로 적용하고, 이후 명시적으로 **.sequential()**을 통해 고유함수 및 정렬을 호출했습니다. 이 하이브리드 접근 방식은 상태 저장된 단말 부분에서만 순차적 병목 현상을 제한하여 70%의 병렬 스루풋을 보존하면서 만남 순서를 보장했습니다.
결과적으로 각 센서 이벤트의 첫 번째 발생을 올바르게 식별하는 안정적이고 결정적인 파이프라인이 구축되었습니다. 처리 속도는 수용 가능한 수준을 유지했고 결함률은 제로로 떨어지며 지연 시간은 운영 한계 내에 머물렀습니다.
왜 forEachOrdered 단말 작업이 병렬 스트림에서 forEach보다 상당히 높은 오버헤드를 발생시키며, 언제 절대적으로 필요합니까?
forEach는 병렬 스레드에서 가용해지는 요소를 협조 없이 처리합니다. 이 접근 방식은 스루풋을 극대화하지만 출력이 스레드 도착 순서로 생성될 수 있습니다. 반면에 forEachOrdered는 원래의 만남 순서를 재구성해야 하며, 이로 인해 프레임워크가 결과를 버퍼링하고 속도가 빠른 스레드가 느린 스레드가 소유한 이전 요소를 기다리도록 할 수 있습니다. 이는 동기화 병목을 생성합니다.
이 작업은 처리의 부작용이 소스 순서를 관찰해야 할 때만 절대적으로 필요합니다. 예를 들어 파일과 같이 위치에 민감한 출력이나 GUI 리스트 모델에 기록하는 것과 같은 경우입니다. 로깅이나 동시 수집에 합산하는 것과 같이 순서에 민감하지 않은 부작용의 경우 forEach가 선호됩니다.
병렬 실행 중 미세한 경합 조건을 방지하기 위해 reduce 연산의 집합 연산자 기능이 요구되며, 이 제약 조건이 위반되면 어떤 일이 발생합니까?
reduce 연산은 스트림을 세그먼트로 분할하고, 각 세그먼트에서 격리된 상태에서 집합 연산자를 적용하여 부분 결과를 생성한 다음, 동일한 집합 연산자(또는 별도의 조합기)를 사용하여 이러한 부분 결과를 결합합니다. 결합법칙은 ((a op b) op c)가 (a op (b op c))와 동일하다는 것을 보장합니다. 이 속성은 요소를 세그먼트로 그룹화하고 부분 결과를 결합하는 순서가 비결정적이고 구현 의존적이기 때문에 요구됩니다.
작업이 비결합적일 경우(예: 위치에 따라 달라지는 구분 기호로 문자열을 연결하는 경우), 병렬 실행은 요소를 순차적으로 실행하는 것과 다르게 그룹화할 수 있습니다. 이는 구분 기호가 뒤섞이거나 비결합적 사용자 정의 숫자 유형에 대한 잘못된 합계와 같은 잘못된 결과를 초래합니다.
고속 호출 잠김 연산인 findFirst와 무한 스트림 간의 특정 상호 작용은 병렬 스트림이 무한정 대기할 수 있는 이유이며, 순차 스트림은 즉시 종료될 수 있는 이유는 무엇입니까?
순차 스트림에서 findFirst는 프레디케이트가 일치하면 즉시 종료될 수 있으며, 이는 무한 스트림에서도 가능합니다. 병렬 스트림에서 프레임워크는 소스를 여러 세그먼트로 분할하여 서로 다른 스레드가 처리합니다. 일치하는 요소가 느린 스레드에 의해 처리되는 세그먼트에 있는 경우, findFirst는 만남 순서를 보장하기 위해 그 스레드가 자신의 세그먼트를 완료할 때까지 기다려야 합니다(또는 요소를 찾기 위해) 다른 세그먼트에 다른 요소가 존재하지 않는지 보장해야 합니다.
스트림이 무정렬이거나 findAny가 사용될 경우, 연산은 일치하는 아무 요소에 대해 즉시 종료되어 메인 스레드가 대기 중인 작업을 취소할 수 있습니다. 후보자들은 종종 정렬된 병렬 무한 스트림에서의 findFirst가 사실상 전역 장벽임을 간과하며, 세그먼트가 일치하기 전에 무한하거나 계산적으로 제한이 없는 경우 교착 상태에 빠질 수 있음을 간과합니다.