理解并行 Java 8 流产生的结果

Question

每次运行并行流以读取和处理文件时，都不会得到相同的结果。

我有关于比萨饼的数据，并希望使用 Map 和全局变量计算不同的变量。我应该只使用全局变量。但是当我运行我的代码时，我每次都会得到不同的结果。输入文件根本没有被修改。

package Assignment;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;

import java.util.stream.Collectors;
import java.util.stream.Stream;

public class GlobalVariables {
    static int vegPizzas = 0;
    static int N_V_Pizzas = 0;
    static int Size_regular = 0;
    static int Size_medium = 0;
    static int Size_large = 0;
    static int Cheese_Burst = 0;
    static int Cheese_regular = 0;
    static int cheap_cheese = 0;

    static Stream<String> reader;

    static int rows = 0;

    public static void main(String[] args) {
        // TODO Auto-generated method stub
        try {
            reader = Files.lines(Paths.get("data/SampleData.csv")).parallel();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

        reader.map(x -> x.split(","))
            .filter(x -> (!(x[0].equals("NULL") || x[1].equals("NULL") || x[2].equals("NULL")     || x[3].equals("NULL"))))
            .map(x -> updateCounts(x)).collect(Collectors.toList());

        printResults();
        System.out.println(rows);//to have a count of rows after filtering of NULL values is     done
    }

    private static void printResults() {
        // TODO Auto-generated method stub
        System.out.println("veg pizzas: " + vegPizzas);
        System.out.println("non veg pizzas: " + N_V_Pizzas);
        System.out.println("regular: " + Size_regular + " medium: " + Size_medium + " large: "    +Size_large);
        System.out.println("cheese burst pizzas: " + Cheese_Burst);
        System.out.println("regular cheese burst: " + Cheese_regular);
        System.out.println("cheaper cheese burst: " + cheap_cheese);
    }

    private static Object updateCounts(String[] x) {
        // TODO Auto-generated method stub
//      static int vegPizzas = 0;
//      static int N_V_Pizzas = 0;
//      static int Size_regular = 0;
//      static int Size_medium = 0;
//      static int Size_large = 0;
//      static int Cheese_Burst = 0;
//      static int Cheese_regular = 0;
//      static int cheap_cheese = 0;
        rows++;
        int flag_regular = 0;
        if(x[9].equals("Y")) {
            vegPizzas++;
        }else if(x[9].equals("N")) {
            N_V_Pizzas++;
        }

        if(x[6].equals("R")) {
            Size_regular++;
            flag_regular = 1;
        }
        else if(x[6].equals("M")) {
            Size_medium++;
        }
        else if(x[6].equals("L")) {
            Size_large++;
        }

        if(x[5].equals("Y")) {
            Cheese_Burst++;
            if(flag_regular == 1) {
                Cheese_regular++;
            }
            if(Integer.parseInt(x[7]) < 500) {
                cheap_cheese++;
            }
        }

        return x;
    }

}

//真实结果或预期结果（每个品种的计数）

蔬菜：5303 非蔬菜：1786 普通：1779 中：2660 大：2650 奶酪爆裂：3499 普通奶酪爆裂：900 便宜的奶酪爆裂：598

//运行-1个结果

蔬菜比萨：5296 非蔬菜比萨：1785 普通：1779 中：2660 大：2649 芝士爆裂披萨：3498 普通奶酪爆裂：900 更便宜的奶酪爆裂：598 7060

//Run-2 结果

蔬菜比萨：5294 非蔬菜披萨：1786 常规：1779 中：2659 大：2648 芝士爆裂披萨：3497 普通奶酪爆裂：900 更便宜的奶酪爆裂：598 7055

//Run-3 结果

蔬菜比萨：5303 非蔬菜披萨：1786 普通：1779 中：2660 大：2650 芝士爆裂披萨：3499 普通奶酪爆裂：900 更便宜的奶酪爆裂：598 7086

我确实经历过这个link。我无法将我的问题与该链接中发布的问题联系起来。我确实注意到，如果我创建一个顺序流，我会得到预期的结果。这方面的任何线索都会有所帮助。

Answer 1

由Stream 管道的map 步骤调用的updateCounts(String[] x) 方法不是线程安全的，它会更新static 变量。

所以当它被多个线程并发调用时，预计每次运行都会产生不同的结果（即static变量的最终值在每次运行中都会不同）。

传递给map 的Function 不应有副作用，尤其是在并行使用Stream 时。

使用Streams 进行此计算的更好方法：

• 创建一个PizzaStatistics 类，将所有原始静态计数器变量作为实例变量。

• updateCounts（应该重命名）将返回一个新的 PizzaStatistics 实例，其中相关计数器设置为 1。它不会更新任何静态变量。

• Stream 管道将使用终端操作 reduce 来生成包含总计的单个 PizzaStatistics 实例。