Java8使用并行流（ParallelSream）

背景

Java8的stream接口极大地减少了for循环写法的复杂性，stream提供了map/reduce/collect等一系列聚合接口，还支持并发操作：parallelStream。

Java8并行流ParallelStream和Stream的区别就是Stream支持行执行，而ParallelStream支持并行执行，提高程序运行效率。Java8的paralleStream用fork/join框架提供了并发执行能力。但是如果使用不当，很容易陷入误区。

paralleStream线程安全吗

package main.stream;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class ParalleStreamTest {
    private static List<Integer> list1 = new ArrayList<>();
    private static List<Integer> list2 = new ArrayList<>();
    private static List<Integer> list3 = new ArrayList<>();
    private static List<Integer> list9 = Collections.synchronizedList(new ArrayList<>());
    private static Lock lock = new ReentrantLock();

    public static void main(String[] args) {
        IntStream.range(0, 10000).forEach(list1::add);

        IntStream.range(0, 10000).parallel().forEach(list2::add);
        IntStream.range(0, 10000).parallel().forEach(list9::add);

        IntStream.range(0, 10000).forEach(i -> {
            lock.lock();
            try {
                list3.add(i);
            }finally {
                lock.unlock();
            }
        });

          //改短代码机上，list2的结果仍然小于10000
//        try {
//            TimeUnit.SECONDS.sleep(10);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
//        List<Integer> list4 = list1.stream().map(item -> item + 1).collect(Collectors.toList());
//        List<Integer> list5 = list1.parallelStream().map(item -> item + 1).collect(Collectors.toList());

        System.out.println("foreach串行执行的大小：" + list1.size());
        System.out.println("foreach并行执行的大小：" + list2.size());
        System.out.println("foreach同步集合并行执行的大小：" + list9.size());
        System.out.println("foreach加锁并行执行的大小：" + list3.size());
//        System.out.println("list4串行加法：" + list4.size());
//        System.out.println("list5并行加法：" + list5.size());

        concurrentFun();
    }

    public static void concurrentFun() {
        List<Integer> list =
                new ArrayList<>();
        for (int i = 0; i <1000; i++) {
            list.add(i);
        }
        List<Integer> parallelStorage = new ArrayList<>() ;
        list
                .parallelStream()
                .filter(i->i%2==0)
                .forEach(i->parallelStorage.add(i));
        System.out.println();

        parallelStorage
                .stream()
                .forEachOrdered(e -> System.out.print(e + " "));

        System.out.println();
        System.out.println("Sleep 5 sec");
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        parallelStorage
                .stream()
                .forEachOrdered(e -> System.out.print(e + " "));
    }
}

执行结果如下：

foreach串行执行的大小：10000
foreach并行执行的大小：7662
foreach同步集合并行执行的大小：10000
foreach加锁并行执行的大小：10000

656 null 312 660 314 562 662 316 664 564 318 null null 566 322 null null 324 570 326 328 676 null 574 332 680 334 578 682 906 336 580 684 338 582 908 686 340 584 910 342 586 912 588 914 590 916 592 918 920 922 924 926 594 596 598 600 602 604 606 608 628 610 928 612 930 614 616 932 934 634 936 636 null 638 620 640 622 624 644 344 646 346 348 876 650 350 878 652 352 880 354 882 356 884 358 532 886 360 534 888 362 536 890 364 538 892 366 540 null null null null null 894 372 718 548 896 374 720 552 554 898 556 722 900 558 902 560 904 728 730 732 734 500 502 968 504 736 506 972 738 510 740 976 512 742 514 978 744 516 980 746 518 982 748 520 984 522 986 524 988 null null 282 992 284 812 688 286 814 690 288 816 692 998 290 818 694 292 820 294 822 938 296 698 824 298 940 700 826 300 942 828 702 302 944 830 704 304 946 832 706 306 948 834 708 782 950 710 952 310 784 712 954 836 786 714 956 838 788 716 958 840 790 842 792 250 962 794 252 964 796 254 966 256 258 260 802 262 804 264 266 806 268 844 808 750 846 810 848 272 850 274 756 852 276 854 278 760 856 280 762 858 764 860 766 862 770 864 772 866 774 868 62 64 438 440 442 444 446 406 448 450 410 452 412 454 414 456 458 418 460 420 462 422 464 424 466 426 428 430 66 434 68 436 70 468 72 470 74 472 76 474 376 78 476 80 478 380 480 84 382 482 86 384 484 88 386 486 388 92 488 490 394 492 396 494 398 496 400 498 402 404 776 94 96 778 98 780 100 102 160 220 32 162 222 34 164 224 108 36 166 110 226 null null null null null null null null null null null null null null null null null null null null null null null null null null null 112 186 56 6 114 8 244 116 60 10 246 118 12 248 120 14 122 126 16 124 128 188 18 130 20 190 132 22 192 134 24 194 26 196 28 198 140 30 200 142 202 144 204 146 206 148 870 208 150 872 210 152 874 212 154 214 216 
Sleep 5 sec
656 null 312 660 314 562 662 316 664 564 318 null null 566 322 null null 324 570 326 328 676 null 574 332 680 334 578 682 906 336 580 684 338 582 908 686 340 584 910 342 586 912 588 914 590 916 592 918 920 922 924 926 594 596 598 600 602 604 606 608 628 610 928 612 930 614 616 932 934 634 936 636 null 638 620 640 622 624 644 344 646 346 348 876 650 350 878 652 352 880 354 882 356 884 358 532 886 360 534 888 362 536 890 364 538 892 366 540 null null null null null 894 372 718 548 896 374 720 552 554 898 556 722 900 558 902 560 904 728 730 732 734 500 502 968 504 736 506 972 738 510 740 976 512 742 514 978 744 516 980 746 518 982 748 520 984 522 986 524 988 null null 282 992 284 812 688 286 814 690 288 816 692 998 290 818 694 292 820 294 822 938 296 698 824 298 940 700 826 300 942 828 702 302 944 830 704 304 946 832 706 306 948 834 708 782 950 710 952 310 784 712 954 836 786 714 956 838 788 716 958 840 790 842 792 250 962 794 252 964 796 254 966 256 258 260 802 262 804 264 266 806 268 844 808 750 846 810 848 272 850 274 756 852 276 854 278 760 856 280 762 858 764 860 766 862 770 864 772 866 774 868 62 64 438 440 442 444 446 406 448 450 410 452 412 454 414 456 458 418 460 420 462 422 464 424 466 426 428 430 66 434 68 436 70 468 72 470 74 472 76 474 376 78 476 80 478 380 480 84 382 482 86 384 484 88 386 486 388 92 488 490 394 492 396 494 398 496 400 498 402 404 776 94 96 778 98 780 100 102 160 220 32 162 222 34 164 224 108 36 166 110 226 null null null null null null null null null null null null null null null null null null null null null null null null null null null 112 186 56 6 114 8 244 116 60 10 246 118 12 248 120 14 122 126 16 124 128 188 18 130 20 190 132 22 192 134 24 194 26 196 28 198 140 30 200 142 202 144 204 146 206 148 870 208 150 872 210 152 874 212 154 214 216

发现问题：

并且每次的结果中并行执行的大小不一致，而串行和加锁后的结果一直都是正确结果。显而易见，stream.parallel.forEach()中执行的操作并非线程安全。list2执行结果不一致，但list4和list5的结果一直是正确的，原因在哪呢。。。。。

最初我以为是因为主线程执行完成后并行流中的线程并未结束，sleep了主线程后发现结果并没有发生改变。

以下方式同样也会出现以上问题

 list.parallelStream()
                .map(e -> {
                    parallelStorage.add(e);
                    return e;
                }).forEachOrdered(e -> System.out.print(e + " "));

首先了解下fork/join框架：

基于“分治”的思想，J.U.C在JDK1.7时引入了一套Fork/Join框架。Fork/Join框架的基本思想就是将一个大任务分解（Fork）成一系列子任务，子任务可以继续往下分解，当多个不同的子任务都执行完成后，可以将它们各自的结果合并（Join）成一个大结果，最终合并成大任务的结果，即第一步是拆分，第二步是分开运算，第三步是合并。这三个步骤分别对应的就是Collector的supplier,accumulator和combiner：

基于以上思想，我们可以认为ArrayList内部维护了一个数组Arr其定义一个变量 n用以表式这个数组的大小那么向这个ArrayList中存储数据的过程可以分解为这么几步：

1.读取数组的长度存入n
2.向这个数组中储入元素arr[n]=a
3.将n+1
4.保存n

而对于list2元素数量不固定的原因就是多线程有可能同时读取到相同的下标n同时赋值，这样就会出现元素缺失的问题了。

解决问题：

1) 将上述转为同步集合， 如上述示例中的list9;

 2) 使用collect和reduce接口，这种收集起来所有元素到新集合是线程安全的，如上述示例中的list5.

 注意：

在使用并行流的时候是无法保证元素的顺序的，也就是即使你用了同步集合也只能保证元素都正确但无法保证其中的顺序。

采用像collect收集元素的方式，不光没有出现Null和数量不一致问题，还排序了！所以，在采用并行流收集元素到集合中时，最好调用collect方法，一定不要采用Foreach方法或者map方法。