Java8 parallelStream机制

Question

有没有办法使用java8 lambda并行处理每个类属性？

例如，对于给定的Car.java：

public class Car{
    private String color;
    private String model;
    private float value;
    private float tax;
    // and others attributes
    .
    .
    .
    // Getters and Setters
}

我想在颜色，型号，价值，税收等方面做一些并行的处理。因此，基本上希望的逻辑是：

parallel processing:
    - color
    - model
    - value
    - tax

then join and update Car.java object instance fields. 

请注意，此问题与cars实例的数量无关。可能只有一个，也可能有多个。

Answer 1

我很好奇，那会是什么样子。假设约束成立，每个属性的处理彼此独立，这将是利用parallelStream()的一种方式。但我非常怀疑，由于并行机器的开销，这是否会带来回报。使用reactive streams，比如RxJava，可以让它变得更加怪异。

public class Car{
    public String color;
    public String model;
    public float value;
    public float tax;

    public Car(String color, String model, float value, float tax) {
        this.color = color;
        this.model = model;
        this.value = value;
        this.tax = tax;
    }

    @Override
    public String toString() {
        return "Car{" +
                "color='" + color + '\'' +
                ", model='" + model + '\'' +
                ", value=" + value +
                ", tax=" + tax +
                '}';
    }
}

@Test
public void process() {
    List<Consumer<Car>> processors = Arrays.asList(
            c -> c.color = printThread(c.color.toLowerCase()),
            c -> c.model = printThread(c.model.toLowerCase()),
            c -> c.value = printThread(c.value * c.value),
            c -> c.tax = printThread(c.tax / c.tax));

    Arrays.asList(new Car("Red", "AlphaGorilla", 1f, 0.5f), new Car("Blue", "Bloated++", 10f, 0.2f))
            .parallelStream().forEach(c -> {
        System.out.println(c);
        processors.parallelStream().forEach(p -> {
            p.accept(c);
            fakeExpensiveComputation();
        });
        System.out.println(c);
    });
}

private <T> T printThread(T smthg) {
    System.out.println(String.format("Calculated value %s in thread %d", smthg.toString(), Thread.currentThread().getId()));
    return smthg;
}

private void fakeExpensiveComputation() {
    try {
        Thread.sleep(3000);
    } catch (InterruptedException e) {
        throw new RuntimeException();
    }
}