java8实战读书笔记：初识Stream、流的基本操做（流计算）

本文是博主在学习《java8实战》的一些学习笔记。

从本节开始，将进入到java8 Stream(流)的学习中来。

本文中的部分示例基于以下场景：餐厅点菜，Dish为餐厅中可提供的菜品，Dish的定义以下：

1public class Dish {
 2    /** 菜品名称 */
 3    private final String name;
 4    /** 是不是素食 */
 5    private final boolean vegetarian;
 6    /** 含卡路里 */
 7    private final int calories;
 8    /** 类型 */
 9    private final Type type;
10
11    public Dish(String name, boolean vegetarian, int calories, Type type) {
12        this.name = name;
13        this.vegetarian = vegetarian;
14        this.calories = calories;
15        this.type = type;
16    }
17
18    public enum Type { MEAT, FISH, OTHER }
19
20    // 省略set get方法
21}

菜单的数据以下：

1List<Dish> menu = Arrays.asList(
 2new Dish("pork", false, 800, Dish.Type.MEAT),
 3new Dish("beef", false, 700, Dish.Type.MEAT),
 4new Dish("chicken", false, 400, Dish.Type.MEAT),
 5new Dish("french fries", true, 530, Dish.Type.OTHER),
 6new Dish("rice", true, 350, Dish.Type.OTHER),
 7new Dish("season fruit", true, 120, Dish.Type.OTHER),
 8new Dish("pizza", true, 550, Dish.Type.OTHER),
 9new Dish("prawns", false, 300, Dish.Type.FISH),
10new Dish("salmon", false, 450, Dish.Type.FISH) );

咱们以一个简单的示例来引入流：从菜单列表中，查找出是素食的菜品，并打印其菜品的名称。

在Java8以前，咱们一般是这样实现该需求的：

1List<String> dishNames = new ArrayList<>();
 2for(Dish d menu) {
 3    if(d.isVegetarian()) {
 4        dishNames.add(d.getName()); 
 5    }
 6}
 7//输出帅选出来的菜品的名称：
 8for(String n : dishNames) {
 9    System.out.println(n);
10}

那在java8中，咱们能够这样写：

1menu.streams() .filter( Dish::isVegetarian).map( Dish::getName) .forEach( a -> System.out.println(a) );

其运行输出的结果：

怎么样，神奇吧！！！

在解释上面的代码以前，咱们先对流作一个理论上的介绍。

流是什么？

流，就是数据流，是元素序列，在Java8中，流的接口定义在 java.util.stream.Stream包中，而且在Collection(集合)接口中新增一个方法：

1default Stream<E> stream() {
2        return StreamSupport.stream(spliterator(), false);
3}

流的简短定义：从支持数据处理操做的源生成的元素序列。例如集合、数组都是支持数据操做的数据结构（容器），均可以作为流的建立源，该定义的核心要素以下：

• 源
  流是从一个源建立来而来，并且这个源是支持数据处理的，例如集合、数组等。
• 元素序列
  流表明一个元素序列（流水线），由于是从根据一个数据处理源而建立得来的。
• 数据处理操做
  流的侧重点并不在数据存储，而在于数据处理，例如示例中的filter、map、forEach等。
• 迭代方式
  流的迭代方式为内部迭代，而集合的迭代方式为外部迭代。例如咱们遍历Collection接口须要用户去作迭代，例如for-each，而后在循环体中写对应的处理代码，这叫外部迭代。相反，Stream库使用内部迭代，咱们只须要对流传入对应的函数便可，表示要作什么就行。

注意：流和迭代器Iterator同样，只能遍历一次，若是要屡次遍历，请建立多个流。

接下来咱们将重点先介绍流的经常使用操做方法。

流的经常使用操做

filter

filter函数的方法声明以下：

1java.util.stream.Stream#filter
2Stream<T> filter(Predicate<? super T> predicate);

该方法接收一个谓词，返回一个流，即filter方法接收的lambda表达式须要知足 （ T -> Boolean ）。

示例：从菜单中选出全部是素食的菜品：

1List<Dish> vegetarianDishs = menu.stream().filter(  Dish::isVegetarian )    // 使用filter过滤流中的菜品。
2                                          .collect(toList())；              // 将流转换成List，该方法将在后面介绍。

舒适提示：流的操做能够分红中间件操做和终端操做。中间操做一般的返回结果仍是流，而且在调用终端操做以前，并不会当即调用，等终端方法调用后，中间操做才会真正触发执行，该示例中的collect方法为终端方法。

咱们类比一下数据库查询操做，除了基本的筛选动做外，还有去重，分页等功能，那java8的流API能支持这些操做吗？
答案固然是确定。

distinct

distinct，相似于数据库中的排重函数，就是对结果集去重。
例若有一个数值numArr = [1,5,8,6,5,2,6]，如今要输出该数值中的全部奇数而且不能重复输出，那该如何实现呢？

1Arrays.stream(numArr).filter(  a -> a % 2 == 0 ).distinict().forEach(System.out::println);

limit

截断流，返回一个i不超过指定元素个数的流。
仍是以上例举例，若是要输出的元素是偶数，不能重复输出，而且只输出1个元素，那又该如何实现呢？

1Arrays.stream(numArr).filter(  a -> a % 2 == 0 ).distinict().limit(1).forEach(System.out::println);

skip

跳过指定元素，返回剩余元素的流，与limit互补。

Map

仍是类比数据库操做，咱们一般能够只选择一个表中的某一列，java8流操做也提供了相似的方法。
例如，咱们须要从菜单中提取全部菜品的名称，在java8中咱们可使用以下代码实现：

1版本1：List<String> dishNames = menu.stream().map( (Dish d) -> d.getName() ).collect(Collectors.toList());
2版本2：List<String> dishNames = menu.stream().map( d -> d.getName() ).collect(Collectors.toList());
3版本3：List<String> dishNames = menu.stream().map(Dish::getName).collect(Collectors.toList());

文章的后续部分尽可能使用最简洁的lambda表达式。

咱们来看一下Stream关于map方法的声明：

1<R> Stream<R> map(Function<? super T, ? extends R> mapper)
2

接受一个函数Function，其函数声明为：T -> R，接收一个T类型的对象，返回一个R类型的对象。

固然，java为了高效的处理基础数据类型（避免装箱、拆箱带来性能损耗）也定义了以下方法：

1IntStream mapToInt(ToIntFunction<? super T> mapper)
2LongStream mapToLong(ToLongFunction<? super T> mapper)
3DoubleStream mapToDouble(ToDoubleFunction<? super T> mapper)

思考题：对于字符数值["Hello","World"] ，输出字符序列，而且去重。
第一次尝试：

1public static void test_flat_map() {
2    String[] strArr = new String[] {"hello", "world"};
3    List<String> strList = Arrays.asList(strArr);
4    strList.stream().map( s -> s.split(""))
5                    .distinct().forEach(System.out::println);
6}

输出结果：

为何会返回两个String[]元素呢？由于map(s -> s.split()) 此时返回的流为Stream，那咱们是否是能够继续对该Steam[String[]],把String[]转换为字符流，其代码以下：

1public static void test_flat_map() {
2    String[] strArr = new String[] {"hello", "world"};
3    List<String> strList = Arrays.asList(strArr);
4    strList.stream().map( s -> s.split(""))
5                    .map(Arrays::stream)
6                    .distinct().forEach(System.out::println);
7}

其返回结果：

仍是不符合预期，其实缘由也很好理解，再次通过map(Arrays:stream)后，返回的结果为Stream，即包含两个元素，每个元素为一个字符流，能够经过以下代码验证：

1public static void test_flat_map() {
 2    String[] strArr = new String[] {"hello", "world"};
 3    List<String> strList = Arrays.asList(strArr);
 4    strList.stream().map( s -> s.split(""))
 5                    .map(Arrays::stream)
 6                    .forEach(  (Stream<String> s) -> {
 7                        System.out.println("\n --start---");
 8                        s.forEach(a -> System.out.print(a + " "));
 9                        System.out.println("\n --end---");
10                    } );
11}

综合上述分析，之因此不符合预期，主要是原数组中的两个字符，通过map后返回的是两个独立的流，那有什么方法将这两个流合并成一个流，而后再进行disinic去重呢？

答案固然是能够的，flatMap方法闪亮登场：先看代码和显示结果：

1public static void test_flat_map() {
2    String[] strArr = new String[] {"hello", "world"};
3    List<String> strList = Arrays.asList(strArr);
4    strList.stream().map( s -> s.split(""))
5                    .flatMap(Arrays::stream)
6                    .distinct().forEach( a -> System.out.print(a +" "));
7}

其输出结果：

符合预期。一言以蔽之，flatMap能够把两个流合并成一个流进行操做。

查找和匹配

Stream API提供了allMatch、anyMatch、noneMatch、findFirst和findAny方法来实现对流中数据的匹配与查找。

allMatch

咱们先看一下该方法的声明：

1boolean allMatch(Predicate<? super T> predicate);

接收一个谓词函数(T->boolean)，返回一个boolean值，是一个终端操做，用于判断流中的全部元素是否与Predicate相匹配，只要其中一个元素不复合，该表达式将返回false。
示例以下：例如存在这样一个List a,其中元素为 1,2,4,6,8。判断流中的元素是否都是偶数。

1boolean result = a.stream().allMatch(  a -> a % 2 == 0 )；  // 将
返回false。

anyMatch

该方法的函数声明以下：

1boolean anyMatch(Predicate<? super T> predicate)
2

一样接收一个谓词Predicate( T -> boolean )，表示只要流中的元素至少一个匹配谓词，即返回真。

示例以下：例如存在这样一个List a,其中元素为 1,2,4,6,8。判断流中的元素是否包含偶数。

1boolean result = a.stream().anyMatch(  a -> a % 2 == 0 )；  // 将返回true。

noneMatch

该方法的函数声明以下：

1boolean noneMatch(Predicate<? super T> predicate);

一样接收一个谓词Predicate( T -> boolean )，表示只要流中的元素所有不匹配谓词表达式，则返回true。

示例以下：例如存在这样一个List a,其中元素为 2,4,6,8。判断流中的全部元素都不式奇数。

1boolean result = a.stream().noneMatch(  a -> a % 2 == 1 )；  // 将返回true。

findFirst

查找流中的一个元素，其函数声明以下：

1Optional<T> findFirst();
返回流中的一个元素。其返回值为Optional，这是jdk8中引入的一个类，俗称值容器类，其主要左右是用来避免值空指针，一种更加优雅的方式来处理null。该类的具体使用将在下一篇详细介绍。

1public static void test_find_first(List<Dish> menu) {
2    Optional<Dish> dish = menu.stream().findFirst();
3    // 这个方法表示，Optional中包含Dish对象，则执行里面的代码，不然什么事不干，是否是比判断是否为null更友好
4    dish.ifPresent(a -> System.out.println(a.getName()));  
5}

findAny

返回流中任意一个元素，其函数声明以下：

1Optional<T> findAny();

reduce

reduce归约，看过大数据的人用过会很是敏感，目前的java8的流操做是否是有点map-reduce的味道，归约，就是对流中全部的元素进行统计分析，归约成一个数值。
首先咱们看一下reduce的函数说明：

1T reduce(T identity, BinaryOperator<T> accumulator)

• T identity：累积器的初始值。
• BinaryOperator< T> accumulator：累积函数。BinaryOperator< T> extend BiFunction。BinaryOperator的函数式表示，接受两个T类型的入参，返回T类型的返回值。

1Optional<T> reduce(BinaryOperator<T> accumulator);

能够理解为没有初始值的归约，若是流为空，则会返回空，故其返回值使用了Optional类来优雅处理null值。

1<U> U reduce(U identity, BiFunction<U, ? super T, U> accumulator, BinaryOperator<U> combiner);
首先，最后的返回值类型为U。

• U identity：累积函数的初始值。
• BiFunction accumulator：累积器函数，对流中的元素使用该累积器进行归约，在具体执行时accumulator.apply( identity, 第二个参数的类型不作限制 )，只要最终返回U便可。
• BinaryOperator< U> combiner：组合器。对累积器的结果进行组合，由于归约reduce，java流计算内部使用了fork-join框架，会对流的中的元素使用并行累积，每一个线程处理流中一部分数据，最后对结果进行组合，得出最终的值。

舒适提示：对流API的学习，一个最最重点的就是要掌握这些函数式编程接口，而后掌握如何使用Lambda表达式进行行为参数化（lambda表达当成参数传入到函数中）。

接下来咱们举例来展现如何使用reduce。
示例1：对集合中的元素求和

1List<Integer> goodsNumber = Arrays.asList(   3, 5, 8, 4, 2, 13 );
2java7以前的示例：
3int sum = 0;
4for(Integer i : goodsNumber) {
5sum += i;//  sum = sum + i;
6}
7System.out.println("sum:" + sum);

求和运算符： c = a + b，也就是接受2个参数，返回一个值，而且这三个值的类型一致。

故咱们可使用T reduce(T identity, BinaryOperator< T> accumulator)来实现咱们的需求：

1public static void test_reduce() {
2    List<Integer> goodsNumber = Arrays.asList(   3, 5, 8, 4, 2, 13 );
3    int sum = goodsNumber.stream().reduce(0, (a,b) -> a + b);
4    //这里也能够写成这样：
5    // int sum = goodsNumber.stream().reduce(0, Integer::sum);
6    System.out.println(sum);
7}

不知你们是否只读(a,b)这两个参数的来源，其实第一个参数为初始值T identity，第二个参数为流中的元素。

那三个参数的reduce函数主要用在什么场景下呢？接下来仍是用求和的例子来展现其使用场景。在java多线程编程模型中，引入了fork-join框架，就是对一个大的任务进行先拆解，用多线程分别并行执行，最终再两两进行合并，得出最终的结果。reduce函数的第三个函数，就是组合这个动做，下面给出并行执行的流式处理示例代码以下：

1 public static void test_reduce_combiner() {
 2
 3    /** 初始化待操做的流 */
 4    List<Integer> nums = new ArrayList<>();
 5    int s = 0;
 6    for(int i = 0; i < 200; i ++) {
 7        nums.add(i);
 8        s = s + i;
 9    }
10
11    // 对流进行归并，求和,这里使用了流的并行执行版本 parallelStream，内部使用Fork-Join框架多线程并行执行，
12    // 关于流的内部高级特性，后续再进行深刻，目前先以掌握其用法为主。
13    int sum2 = nums.parallelStream().reduce(0,Integer::sum, Integer::sum);
14    System.out.println("和为：" + sum2);
15
16    // 下面给出上述版本的debug版本。
17
18    // 累积器执行的次数
19    AtomicInteger accumulatorCount = new AtomicInteger(0);
20
21    // 组合器执行的次数（其实就是内部并行度）
22    AtomicInteger combinerCount = new AtomicInteger(0);
23
24    int sum = nums.parallelStream().reduce(0,(a,b) -> {
25                accumulatorCount.incrementAndGet();
26                return a + b;
27           }, (c,d) -> {
28                combinerCount.incrementAndGet();
29                return  c+d;
30        });
31
32    System.out.println("accumulatorCount:" + accumulatorCount.get());
33    System.out.println("combinerCountCount:" + combinerCount.get());
34}

从结果上能够看出，执行了100次累积动做，但只进行了15次合并。

流的基本操做就介绍到这里，在此总结一下，目前接触到的流操做：

一、filter

• 函数功能：过滤
• 操做类型：中间操做
• 返回类型：Stream
• 函数式接口：Predicate

• 函数描述符：T -> boolean
  二、distinct

• 函数功能：去重
• 操做类型：中间操做

• 返回类型：Stream
  三、skip

• 函数功能：跳过n个元素
• 操做类型：中间操做
• 返回类型：Stream

• 接受参数：long
  四、limit

• 函数功能：截断流，值返回前n个元素的流
• 操做类型：中间操做
• 返回类型：Stream

• 接受参数：long
  五、map

• 函数功能：映射
• 操做类型：中间操做
• 返回类型：Stream
• 函数式接口：Function
• 函数描述符：T -> R

六、flatMap

• 函数功能：扁平化流，将多个流合并成一个流
• 操做类型：中间操做
• 返回类型：Stream
• 函数式接口：Function>
• 函数描述符：T -> Stream
  七、sorted
• 函数功能：排序
• 操做类型：中间操做
• 返回类型：Stream
• 函数式接口：Comparator
• 函数描述符：(T,T) -> int
  八、anyMatch
• 函数功能：流中任意一个匹配则返回true
• 操做类型：终端操做
• 返回类型：boolean
• 函数式接口：Predicate
• 函数描述符：T -> boolean
  九、allMatch
• 函数功能：流中所有元素匹配则返回true
• 操做类型：终端操做
• 返回类型：boolean
• 函数式接口：Predicate
• 函数描述符：T -> boolean
  十、 noneMatch
• 函数功能：流中全部元素都不匹配则返回true
• 操做类型：终端操做
• 返回类型：boolean
• 函数式接口：Predicate
• 函数描述符：T -> boolean
  十一、findAny
• 函数功能：从流中任意返回一个元素
• 操做类型：终端操做
• 返回类型：Optional
  十二、findFirst
• 函数功能：返回流中第一个元素
• 操做类型：终端操做
• 返回类型：Optional
  1三、forEach
• 函数功能：遍历流
• 操做类型：终端操做
• 返回类型：void
• 函数式接口：Consumer
• 函数描述符：T -> void
  1四、collect
• 函数功能：将流进行转换
• 操做类型：终端操做
• 返回类型：R

• 函数式接口：Collector
  1五、reduce

• 函数功能：规约流
• 操做类型：终端操做
• 返回类型：Optional
• 函数式接口：BinaryOperator
• 函数描述符：(T,T) -> T
  1六、count
• 函数功能：返回流中总元素个数
• 操做类型：终端操做
• 返回类型：long
  因为篇幅的缘由，流的基本计算就介绍到这里了，下文还将重点介绍流的建立，数值流与Optional类的使用。

更多文章请关注公众号：

一波广告来袭，做者新书《RocketMQ技术内幕》已出版上市：

《RocketMQ技术内幕》已出版上市，目前可在主流购物平台（京东、天猫等）购买，本书从源码角度深度分析了RocketMQ NameServer、消息发送、消息存储、消息消费、消息过滤、主从同步HA、事务消息；在实战篇重点介绍了RocketMQ运维管理界面与当前支持的39个运维命令；并在附录部分罗列了RocketMQ几乎全部的配置参数。本书获得了RocketMQ创始人、阿里巴巴Messaging开源技术负责人、Linux OpenMessaging 主席的高度承认并做序推荐。目前是国内第一本成体系剖析RocketMQ的书籍。