Java之流编程

实战：集合与流操作对比

分别使用集合操作及Stream流操作，完成对实际应用场景中的数据处理。直观感受流操作带来的便捷性。

购物车案例

沿用Java之函数编程中的购物车的案例。

需求条件：

  1. 想看看购物车中都有什么商品
  2. 图书类商品都给买
  3. 其余的商品中买两件最贵的
  4. 只需要两件商品的名称和总价

传统方式处理业务逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
/**
   * 以原始集合操作实现需求
   */
  public static void oldCartHandle() {
      List<Sku> cartSKuList = CartService.getCartSKuList();
      //1. 打印所有商品
      for (Sku sku : cartSKuList) {
          System.out.println(JSON.toJSONString(sku, true));
      }

      //2. 图书类过滤
      List<Sku> notBooksSkuList = new ArrayList<>();
      for (Sku sku : cartSKuList) {
          if (!SkuCategoryEnum.BOOKS.equals(sku.getSkuCategory())) {
              notBooksSkuList.add(sku);
          }
      }

      //3. 对总价格进行排序
      notBooksSkuList.sort(new Comparator<Sku>() {
          @Override
          public int compare(Sku sku1, Sku sku2) {
              if (sku1.getTotalPrice() > sku2.getTotalPrice()) {
                  return -1;
              } else if (sku1.getTotalPrice() < sku2.getTotalPrice()) {
                  return 1;
              } else {
                  return 0;
              }
          }
      });

      //4. 查找两件最贵的商品
      List<Sku> topTwoSkuList = new ArrayList<>();
      for (int i = 0; i < 2; i++) {
          topTwoSkuList.add(notBooksSkuList.get(i));
      }

      //5. 求两件商品的总价
      Double money = 0.00;
      for (Sku sku : topTwoSkuList) {
          money += sku.getTotalPrice();
      }

      //6. 获取两件商品的名称
      List<String> resultSkuNameList = new ArrayList<>();
      for (Sku sku : topTwoSkuList) {
          resultSkuNameList.add(sku.getSkuName());
      }

      // 打印输出结果
      System.out.println(JSON.toJSONString(resultSkuNameList, true));
      System.out.println("商品总价：" + money);
  }

Lambda、Stream处理业务逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
/**
 * 以Stream流的方式实现需求
 */
public static void newCartHandle() {
    AtomicReference<Double> money = new AtomicReference<>(0.0);

    List<String> resultSkuNameList = CartService.getCartSKuList()
            .stream()
            //1. 打印商品信息
            .peek(sku -> System.out.println(JSON.toJSONString(sku, true)))
            //2. 过滤掉所有图书类商品
            .filter(sku -> !SkuCategoryEnum.BOOKS.equals(sku.getSkuCategory()))
            //3. 根据总价进行排序
            .sorted(Comparator.comparing(Sku::getTotalPrice).reversed())
            //4. 查找两件最贵的商品
            .limit(2)
            //5. 求两件商品的总价
            .peek(sku -> money.set(money.get() + sku.getTotalPrice()))
            //6. 获取两件商品的名称
            .map(Sku::getSkuName)
            // 结果收集
            .collect(Collectors.toList());

    // 打印输出结果
    System.out.println(JSON.toJSONString(resultSkuNameList, true));
    System.out.println("商品总价：" + money.get());
}

Stream流

流是什么？

  • JDK1.8引入的新成员，以声明式方式处理集合数据
  • 将基础操作链接起来，完成复杂的数据处理流水线
  • 提供透明的并行处理

流的简介

从支持数据处理操作的源生成的元素序列。 – Java8实战

  • 元素序列：在java.util.stream.Stream当中定义了一个新的接口，可以访问特定元素类型的一组有序值。当中有stream方法，可以返回一个流。流的目的在于表达计算
  • 源：流会使用一个包含数据的源，比如集合，数组，输入输出的资源等。从有序集合生成流时会保留原有的顺序。由列表生成的流，其元素顺序与列表一致。
  • 数据处理的操作：流的数据处理功能支持类似于数据库的操作，以及函数式编程语言中 的常用操作。流可以顺序执行，也可以并行执行。并行执行用parallStream就行。数据处理的操作就是 一个中间操作链，形成一条流的流水线。这些操作会返回一个流，但是如果没有终端操作，这些操作并不会执行。

流与集合的区别

  • 时间（流）与空间（集合）
    • 流面向于计算
    • 集合面向于存储
  • 只能遍历一次
    • 流只能遍历一次
    • 集合可以遍历多次
  • 外部迭代与内部迭代
    • 流是内部迭代
    • 集合时外部迭代

流的组成

image-20201028115525985

流操作分类

image-20201028115858508

流的使用

image-20201028130214238

  • 有状态：有状态就是有数据存储功能，线程不安全
  • 无状态：无状态就是一次操作，不能保存数据。线程安全

实战案例：Stream操作演示

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
package com.java.example.stream;

import com.alibaba.fastjson.JSON;
import com.java.example.cart.entity.Sku;
import com.java.example.cart.enums.SkuCategoryEnum;
import com.java.example.cart.service.CartService;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

import java.util.*;

/**
 * 演示流的各种操作
 *
 * @author jingLv
 * @date 2020/10/28
 */
public class StreamOperator {

    List<Sku> skus;

    @BeforeClass
    public void init() {
        skus = CartService.getCartSKuList();
    }

    /**
     * filter使用：过滤调不符合断言判断的数据
     */
    @Test
    public void filterTest() {
        skus.stream()
                // 过滤图书类的商品
                .filter(sku -> SkuCategoryEnum.BOOKS.equals(sku.getSkuCategory()))
                .forEach(item -> System.out.println(JSON.toJSONString(item, true)));
    }

    /**
     * map使用：将一个元素转换成另一个元素
     */
    @Test
    public void mapTest() {
        skus.stream()
                // 转换为商品的名称集合
                .map(sku -> sku.getSkuName())
                .forEach(item -> System.out.println(JSON.toJSONString(item, true)));
    }

    /**
     * flatMap使用：将一个对象转换成流
     */
    @Test
    public void flatMapTest() {
        skus.stream()
                // 将商品名称分割为字符流返回
                .flatMap(sku -> Arrays.stream(sku.getSkuName().split("")))
                .forEach(item -> System.out.println(JSON.toJSONString(item, true)));
    }

    /**
     * peek使用：对流进行遍历操作，与forEach类似，但不会销毁流元素
     * peek和forEach都是遍历操作，有什么区别：
     * peek是中间操作，流操作完还是继续可以使用，传递到下一步操作
     * forEach是终端操作，流操作完事不可用的
     * <p>
     * 执行以下代码会发现，peek和forEach交替打印的，并不是peek执行完，在执行forEach，这也是流执行个一个特点：
     * 流是惰性执行，只有遇到终端操作，流的执行才会从上到下依次执行
     */
    @Test
    public void peekTest() {
        skus.stream()
                // sku的名称进行循环打印
                .peek(sku -> System.out.println(sku.getSkuName()))
                .forEach(item -> System.out.println(JSON.toJSONString(item, true)));

    }


    /**
     * sorted使用：对流中元素进行排序，可选择自然排序或指定排序规则。
     * 无参的是按照默认的自然排序操作排
     * <p>
     * 执行以下代码会发现，peek会先打印出来，forEach会后打印出来，这是因为在中间加了一个sorted有状态的操作，所有经过peek的数据都要在sorted里面做一个汇总，由sorted统一排序之后再交由下一个环节处理
     * 有状态操作和无状态的操作区别，会对数据执行的先后有所影响
     */
    @Test
    public void sortedTest() {
        skus.stream()
                .peek(sku -> System.out.println(sku.getSkuName()))
                .sorted(Comparator.comparing(Sku::getTotalPrice))
                .forEach(item -> System.out.println(JSON.toJSONString(item, true)));
    }

    /**
     * distinct操作：对流元素进行去重。
     */
    @Test
    public void distinctTest() {
        skus.stream()
                .map(Sku::getSkuCategory)
                .distinct()
                .forEach(item -> System.out.println(JSON.toJSONString(item, true)));
    }

    /**
     * skip使用：跳过前N条记录
     */
    @Test
    public void skipTest() {
        skus.stream()
                .sorted(Comparator.comparing(Sku::getTotalPrice))
                .skip(3)
                .forEach(item -> System.out.println(JSON.toJSONString(item, true)));
    }

    /**
     * limit使用：截断前N条记录
     */
    @Test
    public void limitTest() {
        skus.stream()
                .sorted(Comparator.comparing(Sku::getTotalPrice))
                .limit(3)
                .forEach(item -> System.out.println(JSON.toJSONString(item, true)));
    }

    /**
     * allMatch使用：终端操作，短路操作。所有元素匹配，返回true
     * 如何查看是短路操作，使用peek打印一下，依次匹配流中数据，只要遇到不匹配则直接返回
     */
    @Test
    public void allMatchTest() {
        boolean b = skus.stream()
                .peek(sku -> System.out.println(sku.getSkuName()))
                .allMatch(sku -> sku.getTotalPrice() > 1000);
        System.out.println(b);
    }

    /**
     * anyMatch使用：任何元素匹配要匹配，返回true
     */
    @Test
    public void anyMatchTest() {
        boolean b = skus.stream()
                .peek(sku -> System.out.println(sku.getSkuName()))
                .anyMatch(sku -> sku.getTotalPrice() > 1000);
        System.out.println(b);
    }

    /**
     * noneMatch使用：任何元素都不匹配返回true
     */
    @Test
    public void noneMatchTest() {
        boolean b = skus.stream()
                .peek(sku -> System.out.println(sku.getSkuName()))
                .noneMatch(sku -> sku.getTotalPrice() > 10000);
        System.out.println(b);
    }

    /**
     * findFirst使用：找到第一个
     */
    @Test
    public void findFirstTest() {
        Optional<Sku> skuOptional = skus.stream().findFirst();
        System.out.println(JSON.toJSONString(skuOptional.get(), true));
    }

    /**
     * findAny使用：找到任何一个
     * findAny和findFirst区别：并行上会有区别，找第一个元素在并行上限制会更多一些，任意元素在并行上就会少，在并行上findAny会比findFirst上快
     * findAny缺点是会随机匹配到元素
     * 流是串行的情况下：findAny和findFirst是没有区别的
     */
    @Test
    public void findAnyTest() {
        Optional<Sku> skuOptional = skus.stream().findAny();
        System.out.println(JSON.toJSONString(skuOptional.get(), true));
    }

    /**
     * max使用：获取流统计的最大值
     */
    @Test
    public void maxTest() {
        OptionalDouble optionalDouble = skus.stream()
                // 获取总价
                .mapToDouble(Sku::getTotalPrice)
                // 获取总价最大的值
                .max();
        System.out.println(optionalDouble.getAsDouble());
    }

    /**
     * min使用：获取流统计的最小值
     */
    @Test
    public void minTest() {
        OptionalDouble optionalDouble = skus.stream()
                // 获取总价
                .mapToDouble(Sku::getTotalPrice)
                // 获取总价最小的值
                .min();
        System.out.println(optionalDouble.getAsDouble());
    }

    /**
     * count使用：获取流中元素的个数
     */
    @Test
    public void countTest() {
        // 获取流中元素的个数
        long count = skus.stream().count();
        System.out.println(count);
    }
}

流的构建

  • 由值创建流
  • 由数组创建流
  • 由文件生成流
  • 由函数生成流（无限流）

实战案例：Stream构建四种形式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
package com.java.example.stream;

import org.testng.annotations.Test;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.Arrays;
import java.util.stream.IntStream;
import java.util.stream.Stream;

/**
 * 流的四种构建形式
 *
 * @author jingLv
 * @date 2020/10/28
 */
public class StreamConstructor {

    /**
     * 由数值直接构建流
     */
    @Test
    public void streamFromValue() {
        Stream<Integer> integerStream = Stream.of(1, 2, 3, 4, 5);
        integerStream.forEach(System.out::println);
    }

    /**
     * 通过数组构建流
     */
    @Test
    public void streamFromArray() {
        int[] numbers = {1, 2, 3, 4, 5};
        IntStream stream = Arrays.stream(numbers);
        stream.forEach(System.out::println);
    }

    /**
     * 通过文件生成流
     *
     * @throws IOException
     */
    @Test
    public void streamFromFile() throws IOException {
        Stream<String> lines = Files.lines(Paths.get("/Users/apple/JavaProject/java-sample/java-new/src/main/resources/Interfaces.sql"));
        lines.forEach(System.out::println);
    }

    /**
     * 通过函数生成的流（无限流）
     */
    @Test
    public void streamFromFunction() {
        // 通过迭代的方式生成流, 无限的生成
        // Stream<Integer> iterate = Stream.iterate(0, n -> n + 2);
        // 通过生成器，随机生成无限的流
        Stream<Double> generate = Stream.generate(Math::random);
        generate.limit(100).forEach(System.out::println);
    }
}

收集器

  • 将流中的元素累积成一个结果
  • 作用域终端操作collect()上
  • collect/Collector/Collectors
    • collect：作为终端操作出现的，流收集的最后一个步骤，一个方法
    • Collector：是一个接口，collect方法需要接收一个实现了Collector接口的收集器才可以收集
    • Collectors：是一个工具类，已经提前封装预制了一些实现了Collector接口的收集器，可以直接哪来用

常用预定义收集器功能

  • 将流元素归约和汇总一个值
  • 将流元素分组
  • 将流元素分区

实战案例：演示预定义收集的使用方式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
package com.java.example.stream;

import com.alibaba.fastjson.JSON;
import com.java.example.cart.entity.Sku;
import com.java.example.cart.service.CartService;
import org.testng.annotations.Test;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 常见预定义收集器使用
 *
 * @author jingLv
 * @date 2020/10/28
 */
public class StreamCollector {

    /**
     * 集合收集器
     */
    @Test
    public void toList() {
        List<Sku> skus = CartService.getCartSKuList();
        List<Sku> collect = skus.stream()
                .filter(sku -> sku.getTotalPrice() > 100)
                .collect(Collectors.toList());
        System.out.println(JSON.toJSONString(collect, true));
    }

    /**
     * 分组
     */
    @Test
    public void group() {
        List<Sku> skus = CartService.getCartSKuList();
        // Map<分组条件, 结果集合>
        Map<Object, List<Sku>> collect = skus.stream().collect(Collectors.groupingBy(Sku::getSkuCategory));
        System.out.println(JSON.toJSONString(collect, true));
    }

    /**
     * 分区
     */
    @Test
    public void partition() {
        List<Sku> skus = CartService.getCartSKuList();
        Map<Boolean, List<Sku>> collect = skus.stream().collect(Collectors.partitioningBy(sku -> sku.getTotalPrice() > 100));
        System.out.println(JSON.toJSONString(collect, true));
    }
}

归约与汇总

  • 归约（reduce）：将Stream流中元素转换成一个值
  • 汇总（collect）：将Stream流中元素转换成一个容器

归约

将Stream流中元素转换成一个值，只返回一个值

1
2
3
4
5
6
7
8
9
10
Stream<Integer> integerStream = Lists.newArrayList(1, 2, 3).stream();

// 求最大值
integerStream.mapToInt(Integer::intValue).max();

// 求最小值
integerStream.mapToInt(Integer::intValue).min();

// 求和
integerStream.mapToInt(Integer::intValue).sum();
归约操作理解

例如：一个Stream流进行累加

image-20201029113847880

reduce接口参数

以reduce最复杂的接口参数为例

1
2
3
4
5
6
7
8
9
10
11
/**
* 归约操作接口定义
* @param identity - 初始值
* @param accumulator - 计算逻辑
* @param combiner - 并行执行时多个部分结果的合并方式
* @param <U> - 元素类型
* @return
*/
<U> U reduce(U identity,
                 BiFunction<U, ? super T, U> accumulator,
                 BinaryOperator<U> combiner);

结合图查看接口做了什么

image-20201029115956187

实战案例：自定义归约

根据一批订单信息，计算平均商品价格

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@Test
  public void reduceTest() {
      /*
       * 订单对象
       */
      @Data
      @AllArgsConstructor
      class Order {
          /**
           * 订单编号
           */
          private Integer id;
          /**
           * 商品数量
           */
          private Integer productCount;
          /**
           * 消费总金额
           */
          private Double totalAmount;
      }

      // 准备数据
      List<Order> list = Lists.newArrayList();
      list.add(new Order(1, 2, 25.12));
      list.add(new Order(2, 5, 257.23));
      list.add(new Order(3, 3, 25512.12));

      // 传统的方式：
      // 1.计算商品数量
      // 2.计算消费总金额

      // 汇总商品数量和总金额
      // 注意：stream().reduce()，Stream直接调用reduce是不会开启并行模式的，加入parallel才会开启并行模式
      Order order = list.stream()
              .parallel()
              .reduce(
                      // 初始化值
                      new Order(0, 0, 0.0),
                      // Stream中两个元素的计算逻辑
                      (Order order1, Order order2) -> {
                          System.out.println("执行计算逻辑");
                          int productCount = order1.getProductCount() + order2.getProductCount();
                          double totalAmount = order1.getTotalAmount() + order2.getTotalAmount();
                          return new Order(0, productCount, totalAmount);
                      },
                      // 并行情况下，多个并行结果如何合并
                      (Order order1, Order order2) -> {
                          System.out.println("执行合并方法");
                          int productCount = order1.getProductCount() + order2.getProductCount();
                          double totalAmount = order1.getTotalAmount() + order2.getTotalAmount();
                          return new Order(0, productCount, totalAmount);
                      }
              );

      System.out.println(JSON.toJSONString(order, true));
  }

汇总

将Stream流中元素转换成一个容器

1
2
3
4
5
6
7
8
9
10
11
Stream<Integer> integerStream = Lists.newArrayList(1, 2, 3).stream();


// 转换成List集合
List<Integer> list = integerStream.collect(Collectors.toList());

// 按奇偶分区
Map<Boolean, List<Integer>> partitions = integerStream.collect(Collectors.partitioningBy(item -> item % 2 == 0));

// 按元素分组
Map<Boolean, List<Integer>> groups = integerStream.collect(Collectors.groupingBy(item -> item));
collect接口参数

三个参数的接口为例

1
2
3
4
5
6
7
8
9
10
11
/**
* 汇总操作接口定义
* @param supplier - 初始化结果容器
* @param accumulator - 添加元素到结果容器逻辑
* @param combiner - 并行执行时多个部分结果容器的合并方式
* @param <R> - 元素类型
* @return
*/
<R> R collect(Supplier<R> supplier,
                  BiConsumer<R, ? super T> accumulator,
                  BiConsumer<R, R> combiner);
实战案例：自定义收集

根据一批订单信息，计算每个用户的平均商品价格

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
@Test
public void collectTest() {
    /*
     * 订单对象
     */
    @Data
    @AllArgsConstructor
    class Order {
        /**
         * 订单编号
         */
        private Integer id;
        /**
         * 用户账号
         */
        private String account;
        /**
         * 商品数量
         */
        private Integer productCount;
        /**
         * 消费总金额
         */
        private Double totalAmount;
    }

    // 准备数据
    List<Order> list = Lists.newArrayList();
    list.add(new Order(1, "zhangsan", 2, 25.12));
    list.add(new Order(2, "zhangsan", 5, 257.23));
    list.add(new Order(3, "lisi", 3, 25512.12));

    // Map<用户账号，订单(商品数量和金额)>
    // 注意：stream().collect()，Stream直接调用collect是不会开启并行模式的，加入parallel才会开启并行模式
    HashMap<String, Order> orderHashMap = list.stream()
            .parallel()
            .collect(
                    () -> {
                        System.out.println("执行初始化容器操作");
                        return new HashMap<String, Order>();
                    },
                    (HashMap<String, Order> map, Order newOrder) -> {
                        System.out.println("执行新元素添加到容器操作");
                        // 1. 新元素的account已经在map中存在了
                        // 2. 新元素的account已经在map中不存在了
                        String account = newOrder.getAccount();
                        // 如果此账号已存在，将新订单的数据累加上
                        if (map.containsKey(account)) {
                            Order order = map.get(account);
                            order.setProductCount(newOrder.getProductCount() + order.getProductCount());
                            order.setTotalAmount(newOrder.getTotalAmount() + order.getTotalAmount());
                        } else {
                            // 如果不存在，直接将新订单存入map
                            map.put(account, newOrder);
                        }
                    },
                    (HashMap<String, Order> map1, HashMap<String, Order> map2) -> {
                        System.out.println("执行并行结果合并操作");
                        map2.forEach((key, value) -> {
                            // merge:如果从map1中发现了这个key，那么就把这个value作为参数，map1中的value也作为参数(order1, order2)进行传递
                            // 如果这个key没有在map1中出现，那直接将value push到map1中
                            map1.merge(key, value, (order1, order2) -> {
                                // 注意：一定要用map1做合并，因为最后collect返回是map1
                                return new Order(0, key,
                                        order1.getProductCount() + order2.getProductCount(),
                                        order1.getTotalAmount() + order2.getTotalAmount());
                            });
                        });
                    }
            );
    System.out.println(JSON.toJSONString(orderHashMap, true));
}

问题：多次初始化容器map会不会存在内容泄露风险

解答：方法里面的局部变量，方法执行完，没有外部引用了，自然会被标记为可回收，等待GC回收。

收集器接口

collect接口参数

1
2
3
4
5
6
7
8
/**
* 汇总操作接口定义
* @param collector - Collector接口实现类
* @param <R>
* @param <A>
* @return
*/
<R, A> R collect(Collector<? super T, A, R> collector);

收集器接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
/**
* Collector 接口
* @param <T> 流中要收集的项目泛型
* @param <A> 累加器的类型，累加器是在收集过程中用于累积部分结果的对象
* @param <R> 收集操作得到的对象的类型
* 例如：public class ToList<T> implements Collector<T, List<T>, List<T>>
*/
public interface Collector<T, A, R> {
		// 建立新的结果容器
    Supplier<A> supplier();

		// 将元素添加到结果容器
    BiConsumer<A, T> accumulator();

    // 合并两个结果容器
    BinaryOperator<A> combiner();

		// 对结果容器应用最终转换
    Function<A, R> finisher();

		// 定义收集器行为，如：是否可以并行，可以使用哪些优化
    Set<Characteristics> characteristics();

实战案例：查找

实现步骤：

  1. 建立数据模型
  2. 初始化数据
  3. 根据已知条件实现

需求：班级中有多名学生，每名学生有3门课的考试成绩。其中缺考科目分数字段为空。需要找出缺考的学生都叫什么名字。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
package com.java.example.stream.cases;

import lombok.AllArgsConstructor;
import lombok.Data;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 案例一
 *
 * @author jingLv
 * @date 2020/10/29
 */
public class CaseOne {

    /**
     * 考试成绩模型
     */
    @Data
    @AllArgsConstructor
    class ExamStudentScore {
        /**
         * 学生姓名
         */
        private String studentName;
        /**
         * 成绩
         */
        private Integer scoreValue;
        /**
         * 科目
         */
        private String subject;
    }

    /**
     * 存放学生考试成绩
     */
    Map<String, List<ExamStudentScore>> stringListMap;

    @BeforeClass
    public void init() {
        stringListMap = new HashMap<>();
        List<ExamStudentScore> zsSourceList = new ArrayList<>();
        zsSourceList.add(new ExamStudentScore("张三", 30, "CHINESE"));
        zsSourceList.add(new ExamStudentScore("张三", 40, "ENGLISH"));
        zsSourceList.add(new ExamStudentScore("张三", 50, "MATHS"));
        stringListMap.put("张三", zsSourceList);

        List<ExamStudentScore> lsSourceList = new ArrayList<>();
        lsSourceList.add(new ExamStudentScore("李四", 80, "CHINESE"));
        lsSourceList.add(new ExamStudentScore("李四", null, "ENGLISH"));
        lsSourceList.add(new ExamStudentScore("李四", 60, "MATHS"));
        stringListMap.put("李四", lsSourceList);

        List<ExamStudentScore> wwSourceList = new ArrayList<>();
        wwSourceList.add(new ExamStudentScore("王五", null, "CHINESE"));
        wwSourceList.add(new ExamStudentScore("王五", null, "ENGLISH"));
        wwSourceList.add(new ExamStudentScore("王五", 70, "MATHS"));
        stringListMap.put("王五", wwSourceList);
    }

    @Test
    public void findStudent() {
        stringListMap.forEach((studentName, scoreList) -> {
            // 匹配score集合中为null,为空返回true，否则为false
            boolean b = scoreList.stream().anyMatch(score -> {
                // anyMatch只要匹配到任意一条符合条件的数据就会返回，不会继续查找了，尽量减少循环的次数，满足需求
                System.out.println(score);
                return score.getScoreValue() == null;
            });

            if (b) {
                System.out.println("此学生[ " + studentName + " ]有缺考");
            }
        });
    }
}

实战案例：去重

需求：标签管理功能模块。允许用户批量添加标签，后台需要对标签去重，并且需要防止数据库中存在同名的标签。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
package com.java.example.stream.cases;

import lombok.AllArgsConstructor;
import lombok.Data;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
import org.testng.collections.Lists;

import java.util.List;

/**
 * 案例二
 *
 * @author jingLv
 * @date 2020/10/29
 */
public class CaseTwo {

    /**
     * 用户请求的创建标签模型
     */
    @Data
    @AllArgsConstructor
    class TagReqDTO {
        /**
         * 标签的名字
         */
        private String name;
        /**
         * 标签值
         */
        private Integer age;
    }

    /**
     * 从DB中查询出来的已经存在的标签名
     */
    List<String> tagListFromDB;
    /**
     * 用户请求的标签列表
     */
    List<TagReqDTO> tagListFromReq;

    @BeforeClass
    public void init() {
        // 数据库中存在的标签名列表
        tagListFromDB = Lists.newArrayList("李四", "王五", "赵六");
        // 用户提交
        tagListFromReq = Lists.newArrayList(
                new TagReqDTO("张三", 10),
                new TagReqDTO("李四", 30),
                new TagReqDTO("王五", 10));
    }

    @Test
    public void distinctTag() {
        tagListFromReq.stream()
                // 里面的表达式为true时，表示这条数据不会被过滤，为false时，没有通过测试，则会被过滤
                .filter(tag -> !tagListFromDB.contains(tag.getName()))
                // 比较元素的equals对元素进行比较
                .distinct()
                .forEach(System.out::println);
    }
}

实战案例：扁平化

需求：权限管理功能模块。查询某用户所有角色下所包含的权限名称。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
package com.java.example.stream.cases;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
import org.testng.collections.Lists;

import java.util.ArrayList;
import java.util.List;

/**
 * 案例三
 *
 * @author jingLv
 * @date 2020/10/29
 */
public class CaseThree {

    /**
     * 角色
     */
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    class Role {
        /**
         * 权限列表
         */
        private List<Permission> permissions;
    }

    /**
     * 权限
     */
    @Data
    @AllArgsConstructor
    class Permission {
        /**
         * 权限名称
         */
        private String name;
    }

    /**
     * 用户权限列表
     */
    List<Role> roles;

    @BeforeClass
    public void init() {
        roles = new ArrayList<>();

        Role adminRole = new Role();
        List<Permission> adminPermissionList = Lists.newArrayList(
                new Permission("删除"),
                new Permission("查看"),
                new Permission("导出")
        );
        adminRole.setPermissions(adminPermissionList);

        Role userRole = new Role();
        List<Permission> userPermissionList = Lists.newArrayList(
                new Permission("新建"),
                new Permission("修改"),
                new Permission("删除"),
                new Permission("查看")
        );
        userRole.setPermissions(userPermissionList);

        roles.add(adminRole);
        roles.add(userRole);
    }

    /**
     * 对两个角色的权限，去并集
     */
    @Test
    public void findPermission() {
        roles.stream()
                // 扁平化Map 获取对象中的集合类属性，组成一个新的流
                // roles是三层结构，角色列表下有两个角色，每个角色下又有权限的列表，扁平化是将这个三层结构的列表只取权限这一层的列表，将权限的列表进行整合，整合为一个新的流，对新的流进行后续操作
                .flatMap(role -> role.getPermissions().stream())
                .peek(permission -> System.out.println("新的流：" + permission))
                .distinct()
                .forEach(System.out::println);
    }
}

实战案例：分组

需求：设计一个对外提供服务的接口，支持调用方传入多个账户编号查询订单。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
package com.java.example.stream.cases;

import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.Data;
import org.testng.annotations.Test;
import org.testng.collections.Lists;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * 案例四
 *
 * @author jingLv
 * @date 2020/10/29
 */
public class CaseFour {

    @Data
    @AllArgsConstructor
    class Order {
        /**
         * 订单编号
         */
        private Integer orderId;
        /**
         * 账户编号
         */
        private String accountId;
    }

    /**
     * 模拟数据库查询
     *
     * @param accountIds 账户id
     * @return 订单列表
     */
    public List<Order> selectFromDB(List<String> accountIds) {
        List<Order> orders = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            orders.add(
                    new Order(i, accountIds.get(i % accountIds.size()))
            );
        }
        return orders;
    }

    /**
     * 接口
     *
     * @param accountIds
     * @return
     */
    public Map<String, List<Order>> queryOrderByAccountIds(List<String> accountIds) {
        return Optional.ofNullable(selectFromDB(accountIds))
                .map(List::stream)
                .orElseGet(Stream::empty)
                // group分组
                .collect(Collectors.groupingBy(Order::getAccountId));
    }

    @Test
    public void test() {
        Map<String, List<Order>> stringListMap = queryOrderByAccountIds(Lists.newArrayList("张三", "李四", "王五"));
        System.out.println(JSON.toJSONString(stringListMap, true));
    }
}

实战案例：排序

需求：在股票中，撮合交易的原则是一段时间内的交易申请，价格越高的先成交；价格一样，下单时间最早的先成交；价格和时间一致，交易量大的先成交；如果价格、时间和交易量都一致，机构优先成交，散户最后成交。

现有一批交易数据，需要确认交易先后顺序。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
package com.java.example.stream.cases;

import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.Data;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 案例五
 *
 * @author jingLv
 * @date 2020/10/29
 */
public class CaseFive {

    /**
     * 交易实体模型
     */
    @Data
    @AllArgsConstructor
    class Trade {
        /**
         * 下单价格
         */
        private BigDecimal price;
        /**
         * 下单时间
         */
        private LocalDateTime time;
        /**
         * 下单量
         */
        private Integer count;
        /**
         * 下单类型：机构/个人
         */
        private String type;
    }

    /**
     * 一段时间内的交易申请
     */
    List<Trade> tradeList;

    @BeforeClass
    public void init() {
        tradeList = new ArrayList<>();

        tradeList.add(new Trade(new BigDecimal(100), LocalDateTime.now().plusSeconds(1), 500, "机构"));
        tradeList.add(new Trade(new BigDecimal(101), LocalDateTime.now().plusSeconds(2), 1, "个人"));
        tradeList.add(new Trade(new BigDecimal(101), LocalDateTime.now().plusSeconds(1), 1, "个人"));
        tradeList.add(new Trade(new BigDecimal(100), LocalDateTime.now().plusSeconds(1), 500, "个人"));
        tradeList.add(new Trade(new BigDecimal(100), LocalDateTime.now().plusSeconds(0), 2, "个人"));
        tradeList.add(new Trade(new BigDecimal(100), LocalDateTime.now().plusSeconds(0), 100, "机构"));
    }

    @Test
    public void sortTrade() {
        System.out.println(JSON.toJSONString(tradeList, true));

        List<Trade> collect = tradeList.stream()
                .sorted(Comparator
                        // 首先按照价格排序，reverseOrder进行排序调整，将自然排序翻转
                        .comparing(Trade::getPrice, Comparator.reverseOrder())
                        // 按照时间先后进行排序，自然排序
                        .thenComparing(Trade::getTime)
                        // 按照交易量排序，自然排序翻转
                        .thenComparing(Trade::getCount, Comparator.reverseOrder())
                        // 自定义排序规则
                        .thenComparing(
                                // 要排序的字段值
                                Trade::getType,
                                // 自定义排序规则
                                (type1, type2) -> {
                                    if ("机构".equals(type1) && "个人".equals(type2)) {
                                        // -1：type1在先， type2在后
                                        return -1;
                                    } else if ("个人".equals(type1) && "机构".equals(type2)) {
                                        return 1;
                                    } else {
                                        return 0;
                                    }
                                }))
                .collect(Collectors.toList());
        System.out.println("排序后的结果！！！");
        System.out.println(JSON.toJSONString(collect, true));
    }
}