Java stream流处理API使用案例

目录

从iteration到Stream操作

Stream创建

filter.map和flatMap方法

提取子Stream和合并Stream

状态变换

Optional

结果的处理

Collecting into Maps

分组

并行Stream

---

       Stream是Java8中，操作集合的一个重要特性。

从iteration到Stream操作

       当你操作一个集合的时候，你通常的做法是迭代每一个元素，然后处理你想要的事情。举个例子：

String contents = new String(Files.readAllBytes(
    Paths.get("alice.txt")), StandardCharsets.UTF_8);  // 读取文件到字符串中
List<String> words = Arrays.asList(contents.split("[\\P{L}]+"));  // 截取words

       现在我们来迭代操作它：

int count=0;
for (String w : words) {
    if (w.length() > 12) count++;
}

       这段代码有什么问题吗？除了并行处理不是很好以为我想是没有。在Java8中，相同的操作是这么处理的：

long count = words.stream().filter(w -> w.length() > 12).count();

       so cool！从代码中，我们就能非常容易的看出它要表达的意思，filter是针对words的过滤。

       有人会问，这样的操作的确让人很是兴奋，但是刚才说到的并行处理它能解决吗？答案是不可以，但是Java8给我们提供了非常好的API，并行处理如下：

long count = words.parallelStream().filter(w -> w.length() > 12).count();

       通过将stream()修改为parallelStream()，这样就可以并行的进行过滤和统计了。

       从表面上，stream看起来和集合很像，你可以*的操作它。但是有以下几点不同：

1. stream不存储集合元素
2. stream操作不修改源数据，他们是返回一个新的streams来承载结果
3. stream操作都会尽可能的进行延迟加载。这意味着当需要使用结果的时候它才会才运行。

       当你使用stream的时候，需要关注一下3个阶段：

1. 创建一个stream
2. 指定的中间操作将初始stream转化为其他stream
3. 最终的操作会产生一个结果，在调用最终操作前都会延迟执行的。在这之后，stream不会再被使用。

       在上面的例子中，stream()和parallelStream()创造了stream，filter方法转化了它，count方法是最终的操作。

       注意：stream操作不是按照引用的顺序来执行的。在例子中，知道count调用才执行。当count方法请求第一个元素，filter方法开始请求元素，知道找到一个长度大于12的元素。 

Stream创建

       Java8中，你就可以将集合的操作都利用stream来处理，如果你有一个数组，使用静态的Stream.of方法来实现：

// 分隔后返回String[]
Stream<String> words = Stream.of(content.split("[\\P{L}]+"));

       of方法定义：

public static<T> Stream<T> of(T t);
public static<T> Stream<T> of(T... values);

       所以你可以构造很多参数的stream。

       使用Arrays.stream(array, from, to)，可以得到数组的一部分。

       想要创建一个空的stream，可以使用Stream.empty方法：

Stream<String> silence = Stream.empty();

       stream接口有2个静态方法来构造无限的streams。generate方法是通过无参函数的。比如：

Stream<String> echos = Stream.generate(() -> "Echo");
or
Stream<Double> randoms = Stream.generate(Math::random);

       如果想要产生一个无限的序列，例如：0 1 2 3 ...，可以使用iterate方法来实现：

Stream<BigInteger> integers = Stream.iterate(BigInteger.ZERO, n -> n.add(BigInteger.ONE));

filter.map和flatMap方法

 

       map方法也可以达到上面例子的效果，例如：

Stream<String> lowercaseWords = words.map(String::toLowerCase);

       使用Lambda表达式如下：

Stream<Character> firstChars = words.map(s -> s.charAt(0));

       上面这段代码是将words中每个词的第一个字符取出来。

       在Java8之前，如果你想将一段字符串分隔出来，代码如下：

public static Stream<Character> characterStream(String s) {
    List<Character> result = new ArrayList<>();
    for (char c : s.toCharArray()) result.add(c);
    return result.stream();
}

       调用characterStream("boat")， 将返回['b','o','a','t']

       如果用map来操作的话，如下：

Stream<Stream<Character>> result = words.map(w -> characterStream(w));

       这里也可以使用flatMap来替代map：

Stream<Character> letters = words.flatMap(w -> characterStream(w));

提取子Stream和合并Stream

stream.limit(n)方法和SQL中的limit很像，就是取前n个符合条件的数据，例如：

List<Person> persons = Arrays.asList(
        new Person("安红", "女", 12),
        new Person("索隆", "男", 23),
        new Person("路飞", "男", 22)
);

Stream<Person> partialPersonStream = persons.stream().limit(2);
// 将Stream转换为List
List<Person> partialPerson = partialPersonStream.collect(Collectors.toList());
Assert.assertTrue(2 == partialPerson.size());
Assert.assertTrue("安红".equals(partialPerson.get(0).getName()));
Assert.assertTrue("索隆".equals(partialPerson.get(1).getName()));

       stream.skip(n)为跳过前n个数据，例如：

List<Person> persons = Arrays.asList(
        new Person("安红", "女", 12),
        new Person("索隆", "男", 23),
        new Person("路飞", "男", 22)
);

// 跳过前2个
Stream<Person> skipPersonStream = persons.stream().skip(2);
List<Person> skipPersons = skipPersonStream.collect(Collectors.toList());
Assert.assertTrue(1 == skipPersons.size());
Assert.assertTrue("路飞".equals(skipPersons.get(0).getName()));

       stream.concat()方法可以对stream进行追加操作，例如：

Stream<String> combined = Stream.concat(characterStream("Hello"), characterStream("world"));

       这里stream的结果为['h','e','l','l','o','w','o','r','l','d']

       也许大家会觉得这样的操作，一旦debug起来会比较困难，但是大家不用着急，peek方法就会很好的解决这个问题了：、

Object[] powers = Stream.iterate(1.0, p->p*2).peek(e->System.out.println(e)).limit(20).toArray();

       这样就可以将每次得到的结果打印出来了，怎么样，Java8更干、更爽、更贴心吧？

状态变换

       distinct方法用来获取唯一性数据，清除重复数据

// distinct - 清除重复数据
Stream<String> uniqueWordStream = Stream.of("a", "b", "b", "c", "a", "d").distinct();
List<String> uniqueWords = uniqueWordStream.collect(Collectors.toList());
Assert.assertTrue(4 == uniqueWords.size());  // a, b, c, d

Optional

       max和min分别返回最大值和最小值，例如：

Optional<String> largest = words.max(String::compareToIgnoreCase);
if (largest.isPresent()) {
    System.out.println("largest: " + largest.get());
}

// max - 获取最大值
Optional<Person> maxAgePerson = persons.stream().max((p1, p2) -> Integer.compare(p1.getAge(), p2.getAge()));
if (maxAgePerson.isPresent()) {
    Person person = maxAgePerson.get();
    Assert.assertTrue("索隆".equals(person.getName()));
}

       findFirst会返回第一个数据：

// 返回Q开头的第一个数据
Optional<String> startWithQ = words.filter(s->s.startWith("Q")).findFirst();

       如果想匹配所有符合要求的，可以使用如下：

// 匹配所有Q开头的字符串
Optional<String> startsWithQ = words.parallel().filter(s->s.startWith("Q")).findAny();

       如果想要查找是否存在匹配，可以如下：

boolean isWordStartWithQ = words.parallel().anyMatch(s->s.startWith("Q"));

       Optional对象是可以是一个T的包装类，也可以不是对象。它主要是为了替换引用类型T对象或者null。正确使用的话会更安全。

结果的处理

       结果处理使用collect来处理，例子如下：

// stream的结果收集和处理
Stream<Person> stream = persons.stream().filter(p->p.getAge() > 20);
Set<Person> setResult = stream.collect(Collectors.toSet());
// 获取list中name一列的数据
Stream<String> nameStream = persons.stream().map(p->p.getName());
//Stream<String> nameStream = stream.map(p->p.getName());  // Error: 不要这么操作，否则会有问题，无法找到p
List<String> names = nameStream.collect(Collectors.toList());
Assert.assertTrue(3 == names.size());
// 将所有名字连接起来，逗号分隔
String nameStr = nameStream.collect(Collectors.joining(","));  // Error: 执行到这里的时候就会报错了，因为nameStream已经关闭了
Assert.assertTrue("安红,索隆,路飞".equals(nameStr));

Collecting into Maps

       toMap用来处理将数据转化为map结构，例子如下：

// Colloecting into Maps
Map<String, Integer> peoples = persons.stream().collect(Collectors.toMap(Person::getName, Person::getAge));

Stream<Locale> locales = Stream.of(Locale.getAvailableLocales());
Map<String, String> languageNames = locales.collect(
    Collectors.toMap(
        l -> l.getDisplayLanguage(),
        l -> l.getDisplayLanguage(l),
        (existingValue, newValue) -> existingValue));

分组

       分组使用函数groupingBy，例子如下：

Stream<Locale> locales1 = Stream.of(Locale.getAvailableLocales());
Map<String, List<Locale>> countryToLocales = locales1.collect(Collectors.groupingBy(Locale::getCountry));
List<Locale> swissLocales = countryToLocales.get("CH");

并行Stream

       例子如下：

// parallel stream
ConcurrentMap<String, List<Person>> result = persons.stream().parallel().collect(Collectors.groupingByConcurrent(Person::getName));

 

备注：原文地址