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

#java #java_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();


Проблема в том, что не компилируется.. 
    

Ответы

Ответ 1

Если я правильно понял вашу проблему, то надо делать как-то так:

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.