使用 parallelStream 和 forEach 的 Java 并行化，其中每个部分都是完全独立的答案

【问题标题】：Java parallelization using parallelStream and forEach where each part is completely separate使用 parallelStream 和 forEach 的 Java 并行化，其中每个部分都是完全独立的
【发布时间】：2017-07-17 10:54:56
【问题描述】：

我正在编写执行以下操作的（在线机器学习）代码：

你有数据和标签（数据中有N个元素）
有K个基分类器：classifiers[1..K]
对于每个元素（连续：对于t = 1到N，因为这是在线学习），将数据中的第t个元素转换为new_data_elem[1..K]。这是棘手的部分，我决定使用 parallelStream()
然后对于每个第 j 个分类器，确定标签：labels[1..K]（也可以并行）
将labels[1..K]的结果投票到单个dedicated_label中，并与true_label进行比较
然后对于 j = 1:K，您应用分类器[j].update(new_data_elem[j],decided_label == true_label)（更新需要知道预测是否正确）（再次，并行）

这是我的（伪）代码。我已经检查并进行了基准测试，它得到了与顺序版本相同的结果，并且我已经看到使用它有相当大的加速，但我不确定它是否存在潜在的错误：

Set<Integer> set_of_index = new HashSet<>();
for (int j = 0; j < K; j++){
        set_of_index.add(j); // For parallelization 
}  
for (int t = 0; t < n; t++){
        true_label = true_labels[t];
        ... // make new_data_elem[1..K] 
        set_of_index.parallelStream().forEach((j)->{ 
              // Predict
              labels[j] = classifiers[j].predict(new_data_elem[j]);
        }
        ... // using labels[j] to predict decided_label        
        set_of_index.parallelstream().forEach((j)->{
              // Update
              classifiers[j].update((new_data_elem[j],decided_label == true_label);
        }
}

请检查这是否确实正确，因为我读过：http://docs.oracle.com/javase/8/docs/api/java/util/stream/package-summary.html，上面写着

少量的流操作，例如 forEach() 和 peek()，只能通过副作用进行操作；这些应小心使用。

，所以我不太确定:(

【问题讨论】：

标签： java algorithm parallel-processing java-stream

【解决方案1】：

修改数组元素不会干扰对不同数组元素的并发修改，因此，假设您的计算不干扰，代码似乎是正确的。

但有专用的 API 方法使代码更加简洁：

for (int t = 0; t < n; t++){
        true_label = true_labels[t];
        // make new_data_elem[1..K] 
        Arrays.parallelSetAll(labels, j -> classifiers[j].predict(new_data_elem[j]));
        // using labels[j] to predict decided_label        
        IntStream.range(0, K).parallel().forEach( j ->
            classifiers[j].update(new_data_elem[j], decided_label == true_label));
}

Arrays.parallelSetAll 旨在写入每个数组元素，并且更容易推断这些数组写入之间不会发生干扰。通过使用IntStream.range(0, K)，您完全摆脱了set_of_index。

【讨论】：