java8新特性之数据流（串行流和并行流）

1.什么是数据流（what）

java.util.Stream表示了某一种元素的序列，在这些元素上可以进行各种操作。

2.解决什么问题（why）

Stream操作可以是中间操作，也可以是完结操作。完结操作会返回一个某种类型的

值，而中间操作会返回流对象本身，并且你可以通过多次调用同一个流操作方法来将操作结果串起来！

3.怎么用（waht do）

3.1 语法

注：

中间操作：返回的结果是还是Stream流，允许在其返回结果上继续进行流的操作
中止操作：不返回流，所以我们不能再调用其他的流操作

所有的demo都会在3.2

（大家生理

1. Filter（中间操作）

接受一个Boolean类型的变量，并将所有流对象中的元素进行过滤

3. Sorted

Sorted是一个中间操作，能够返回一个排过序的流对象的视图。可指定排序规则!默认正序，逆序：orted(Comparator.reverseOrder())（注意：他的排序是按照ASCII码值排序）

4. Map

map是一个对于流对象的中间操作，通过给定的方法，它能够把流对象中的每一个元素对应到另外一个对象上。

5. Match

所有的匹配操作都是终结操作。匹配操作有多种不同的类型，都是用来判断某一种规则是否与流对象相互吻合的。只返回一个boolean类型的结果。有anyMatch、allMatch、noneMatch。

6. Count

Count是一个终结操作，它的作用是返回一个数值，用来标识当前流对象中包含的元素数量。

7. Reduce

Reduce操作是一个终结操作，它能够通过某一个方法，对元素进行削减操作。该操作的结果会放在一个Optional变量里返回。

8. ForEach

ForEach是一个中止操作。ForEach接受一个function接口类型的变量，用来执行对每一个元素的操作。。它不返回流，所以我们不能再调用其他的流操作。

3.2 示例

//通过串行流实现过滤判断。
        List<String> strings = Arrays.asList("abc", "", "bc", "efg", "abed", "", "jkl");
        //Tips limit获取指定数量的流
        List<String> filtered = strings.stream().limit(3).filter(string -> !string.isEmpty()).collect(Collectors.toList());
        filtered.forEach(System.out::println);
        System.out.println("====================");
        //通过并行流实现迭代
        List<String> parallelString = strings.parallelStream().filter(s -> !s.isEmpty()).collect(Collectors.toList());
        parallelString.forEach(System.out::println);

        //forEach来迭代流中的每个数据
        System.out.println("forEach来迭代流中的每个数据");
        parallelString.forEach(s->{
            s="er";
            System.out.println(s);
        });


        //map 方法用于映射每个元素到对应的结果，也可以用于迭代流中的元素
        List<Integer> numbers = Arrays.asList(3, 2, 2, 3, 7, 3, 5);
        //Tips: distinct:去重作用
        List<Integer> squaresList = numbers.stream().map(i -> i * i).distinct().collect(Collectors.toList());
        squaresList.forEach(System.out::println);

        List<String> stringList = parallelString.stream().map(s -> s + s).distinct().collect(Collectors.toList());
        stringList.forEach(System.out::println);

        //filter 方法用于通过设置的条件过滤出元素。以下代码片段使用 filter 方法过滤出空字符串：
        List<String> filterString = Arrays.asList("abc", "", "bc", "efg", "abed", "", "jkl");
        // 获取空字符串的数量
        long count = filterString.stream().filter(String::isEmpty).count();
        List<String> filterCollect = filterString.stream().filter(String::isEmpty).collect(Collectors.toList());
        System.out.println(count);
        System.out.println(filterCollect.size());

        //sorted 方法用于对流进行排序。以下代码片段使用 sorted 方法对输出的 10 个随机数进行排序!
        Random random = new Random();
        random.ints().limit(10).sorted().forEach(System.out::println);

        //Collectors 类实现了很多归约操作，例如将流转换成集合和聚合元素。Collectors 可用于返回列表或字符串
        List<String> collectList = strings.stream().filter(string -> !string.isEmpty()).collect(Collectors.toList());
        System.out.println("筛选列表: " + collectList);

        String mergedString = strings.stream().filter(string -> !string.isEmpty()).collect(Collectors.joining(""));
        System.out.println("合并字符串: " + mergedString);

        List<Integer> integerList = Arrays.asList(3, 2, 2, 3, 7, 3, 5);
        //一些产生统计结果的收集器也非常有用。它们主要用于int、double、long等基本类型上，它们可以用来产生类似如下的统计结果。
        IntSummaryStatistics stats = integerList.stream().mapToInt((x) -> x).summaryStatistics();
        System.out.println("列表中最大的数 : " + stats.getMax());
        System.out.println("列表中最小的数 : " + stats.getMin());
        System.out.println("所有数之和 : " + stats.getSum());
        System.out.println("平均数 : " + stats.getAverage());

4.并行流

4.1并行流操作

流操作可以是顺序的，也可以是并行的。顺序操作通过单线程执行，而并行操作则通过多线程执行。

package com.jimi.java8.streamapi;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @author test
 */
public class ParallelStreamAndStream {
    public static void main(String[] args) {
        List<Integer> i = Arrays.asList(1, 2, 3, 4, 5, 6, 7);
        System.out.println("stream.forEach打印结果如下(有序)");
        i.stream().forEach(System.out::println);
        System.out.println("parallelStream.forEach打印结果如下(无序)");
        i.parallelStream().forEach(System.out::println);
        System.out.println("parallelStream.forEachOrdered打印结果如下(有序.结果同stream.forEac)");
        i.parallelStream().forEachOrdered(System.out::println);


        //模拟10000条数据 循环打印测试
        List<Integer> list = new ArrayList();
        for (int j = 0; j < 1000; j++) {
            list.add(j);
        }
        // 统计并行执行list的线程
        Set<Thread> threadSet = new CopyOnWriteArraySet<>();
        // 并行执行
        list.parallelStream().forEach(integer -> {
            Thread thread = Thread.currentThread();
            // 统计并行执行list的线程
            threadSet.add(thread);
        });
        System.out.println(threadSet);
        //打印结果由此证明parallelStream是多管道线程
        System.out.println("threadSet一共有" + threadSet.size() + "个线程");
        System.out.println("系统一个有" + Runtime.getRuntime().availableProcessors() + "个cpu");

        list.parallelStream().forEach(System.out::println);


    }
}

4.2 并行流效率测试

package com.jimi.java8.streamapi;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.TimeUnit;

/**
 * @author test
 */
public class ParallelTime {

    public static void main(String[] args) throws InterruptedException {

        //模拟10000条数据 forEach打印测试
        List<Integer> list = new ArrayList();
        for (int j = 0; j < 10; j++) {
            list.add(j);
        }


        long startTime = System.currentTimeMillis();
        //传统for循环
        for (int i = 0; i < list.size(); i++) {
            try {
                //睡眠1毫秒
                TimeUnit.MILLISECONDS.sleep(1);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        long endTime = System.currentTimeMillis();
        System.out.println("传统for循环运行时间:" + (endTime - startTime) + "ms");

        // 测试单管道stream执行效率
        startTime = System.currentTimeMillis();
        list.stream().forEach(r -> {
            try {
                //睡眠1毫秒
                TimeUnit.MILLISECONDS.sleep(1);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        long streamendTime = System.currentTimeMillis();
        System.out.println("stream : " + (streamendTime - startTime) + "ms");

        // 测试多管道parallelStream执行效率
        startTime = System.currentTimeMillis();
        Set<Thread>  threads = new CopyOnWriteArraySet<>();
        list.parallelStream().forEach(r -> {
            try {
                Thread thread = Thread.currentThread();
                threads.add(thread);
                System.out.println("系统创建了"+threads.size()+"个线程");
                //睡眠1毫秒
                TimeUnit.MILLISECONDS.sleep(1);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        long parallelStreamendTime = System.currentTimeMillis();
        System.out.println("parallelStream : " + (parallelStreamendTime - startTime) + "ms");


    }
}

并行流快了很多。但是我产生了另外的一个问题便是：有时候在排查错误的时候，流的形式不方便进行日志的监控。

java8新特性之数据流（串行流和并行流）

4.3 最后啰嗦一句

ConcurrentModificationException总是会在使用迭代器且直接操作集合时产生

java8新特性之数据流（串行流和并行流）

1.什么是数据流（what）

2.解决什么问题（why）

3.怎么用（waht do）

3.1 语法

3.2 示例

4.并行流

4.1并行流操作

4.2 并行流效率测试

4.3 最后啰嗦一句

继续阅读

关于Gradle配置的小结

Java小案例——随机数猜测随机数猜测

nginx location中斜线的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的简单使用

neo4j之cypher使用文档

GitHub连夜封杀！这份阿里 10W 字内部 Java 字面试手册到底有多强？

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结

mybatis_入门程序Mybatis入门

AOP编程_Android优雅权限框架(1)概念基础，2021金三银四前言正文大纲正文

Effective Java 8:通用程序设计

OOM三种类型

工厂模式-三种类型

【递归】高效率求2的n次幂

win10本地scala和spark安装安装scala安装spark

scala (3) Function 和 Method