为什么这个流不能并行运行？答案

【问题标题】：Why can't this stream run in parallel?为什么这个流不能并行运行？
【发布时间】：2016-06-05 19:05:25
【问题描述】：

如何让它并行运行？我的输出是完全连续的，并且在一切完成之前不会显示经过的时间。这是我第一次尝试并行化流，所以我可能遗漏了一些对其他人来说显而易见的东西。但与我查看的示例代码相比，我看不出有什么问题。

public class Paralells {


    void run() {
        for (int i = 1; i<=1000; i++) System.out.println(i);
    }


    public static void main(String[] args) {

        ArrayList<Paralells> a = new ArrayList();

        a.add(new Paralells());
        a.add(new Paralells());
        a.add(new Paralells());
        a.add(new Paralells());
        a.add(new Paralells());

        long x = System.currentTimeMillis();
        a.parallelStream().forEach(p -> p.run());
        System.out.println("ELAPSED: " + (System.currentTimeMillis()-x));
    }

}

【问题讨论】：

您将并行流与异步执行混淆了。 a.parallelStream().forEach(p -> p.run()); 不是异步执行的，它会等待操作完成后再打印"ELAPSED: ..."。但是管道本身是并行运行的； forEach 会被不同的线程同时调用，例如尝试打印当前线程，你会看到它发生了变化。
现在我知道管道是并行的，但是直到管道完成后才会执行下一条语句，这让我有些困惑......

标签： java parallel-processing java-stream

【解决方案1】：

但它是平行的！试验一下：

import java.util.*;
public class Paralells {

    private int id;
    public Paralells(int id) { this.id = id; }
    void run() {
        for (int i = 1; i<=1000; i++) System.out.println(id+" "+i);
    }


    public static void main(String[] args) {

        ArrayList<Paralells> a = new ArrayList();

        a.add(new Paralells(1));
        a.add(new Paralells(2));
        a.add(new Paralells(3));
        a.add(new Paralells(4));
        a.add(new Paralells(5));

        long x = System.currentTimeMillis();
        a.parallelStream().forEach(p -> p.run());
        //a.forEach(p -> p.run()); // sequential
        System.out.println("ELAPSED: " + (System.currentTimeMillis()-x));
    }
}

为每个元素设置一个 id，并在打印消息时使用。您应该能够观察到消息是交错的。与顺序版本比较（取消注释相应的行并注释并行版本）。

当然，时间是在一切完成后收集的。时间在forEach之后打印...

【讨论】：

【解决方案2】：

我尝试了您的确切代码，它确实并行执行。我认为它可能看起来是连续的，因为run() 方法完成得太快，以至于流没有时间创建所有线程。

如果您想查看流实际上是并行运行的，请尝试在 run() 方法的循环中添加一点 Thread.sleep(10) 调用。

现在，对于第二部分，在一切完成之前不会打印经过的时间。即使流是并行的，它仍然会阻塞。

【讨论】：

啊，我明白了，我的观察力不是那么好……发生的事情是，在“切换”输出到另一个线程之前，为该线程吐出了 10-20 个数字。所以它看起来是连续的。

【解决方案3】：

import java.util.ArrayList;

public class Parallels implements Runnable {

    public void run() {
        for (int i = 1; i<=1000; i++) {
// uncomment this line, if you don't see concurrent execution
//            try { Thread.sleep(1); } catch (Exception e) {} 
            System.out.println(i);
    }   }


    public static void main(String[] args) {

        ArrayList<Thread> a = new ArrayList<>();

        for(int i=0; i<5; i++) a.add(new Thread(new Parallels()));

        long x = System.currentTimeMillis();

        for(Thread t : a) t.start();

        for(Thread t : a) try { 
            t.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("ELAPSED: " + (System.currentTimeMillis()-x));
    }
}

实现 Runnable 接口
使用 new Thread(new Parallels()) 创建线程
使用 t.start() 开始他们的执行
使用 t.join() 等待线程完成
如果您没有看到任何并发，请插入一个 Thread.sleep()，它会减慢线程并“激励”VM 切换到另一个线程

【讨论】：