Как распараллелить вложенные циклы с помощью Stream

Познавая Java 8 открыл для себя стримы. Решил попробовать распараллелить достаточно долгие вложенные циклы, но столкнулся с проблемой - пока не знаю как.
Вот так выглядит последовательный код:
long total = 0; BitVector vector = getVector(); // из библиотеки import cern.colt.bitvector.* for (int index = 0; index < vector.size(); index++){ if (vector.get(index)){ for (TrafficData trafficData : trafficDataList){
int visitorTrafficDataIndex = getVisitorTrafficDataIndex(index, trafficData);
if (visitorTrafficDataIndex >= 0){ totalTrafficImpressions += (long)(trafficData.getVolume()[visitorTrafficDataIndex]); } } } }
Думал сделать что-то типа такого:
long total = IntStream.range(0, vector.size()).parallel() .filter(index -> vector.get(index)) .mapToLong(index -> trafficDataList.stream() .map(trafficData -> getVisitorTrafficDataIndex(index, trafficData)) .filter(visitorTrafficDataIndex -> visitorTrafficDataIndex >= 0) .mapToLong(visitorTrafficDataIndex -> (long)(trafficData.getVolume()[visitorTrafficDataIndex]) // здесь уже нет объекта использовать trafficData .sum()) ).sum();
Проблема в том, что не компилируется..

Ответ

Если я правильно понял вашу проблему, то надо делать как-то так:
long total = IntStream.range( 0, vector.size() ).parallel().filter( index -> vector.get( index ) ) .mapToLong( index -> trafficDataList.stream().parallel().map( trafficData -> { int visitorTrafficDataIndex = getVisitorTrafficDataIndex( index, trafficData ); if (visitorTrafficDataIndex >= 0) { return trafficData.getVolume()[visitorTrafficDataIndex]; } return 0; } ).sum() ).sum();
Сделать так как хотите вы (всё на стримах) не получится, потому что после того как вы map(trafficData -> getVisitorTrafficDataIndex(index, trafficData)) в вашем стриме уже нет объекта trafficData, он(стрим) уже просто коллекция int.