Java8 Stream简介

Stream是Java 8新增的重要特性, 它提供函数式编程支持并允许以管道方式操作集合. 流操作会遍历数据源, 使用管道式操作处理数据后生成结果集合, 这个过程通常不会对数据源造成影响.

函数对象

使用Stream进行函数式编程时经常需要将操作作为参数传入流方法中, 函数对象即将方法或lambda表达式作为对象.

import java.util.stream.Collectors;
List list = Arrays.asList(-1,0,1,2,3).stream()
				.filter(x -> x>0)
				.collect(Collectors.toList());

上述示例中filter的参数x -> x>0即为一个lambda表达式.

lambda表达式语法通常为(args)->{body}, 返回值的类型自动推定:

(int a, int b) -> { 
	if (a>b) {
		return a+b;
	} 
	else {
		return a * b;
	}
}

参数的类型也可以自动推定:

(a, b) -> {
  return a+b;
}

在只有一条语句的情况下{}可省略, 返回值类型与语句主体表达式一致.

(a, b) -> a+b

允许使用空参数:

() -> {System.out.println("Hello World!")}

::运算符用于将方法表示为函数对象:

List<Item> items = new ArrayList<>();
items.add(new Item("a"));
items.add(new Item("b"));
List<String> list = items.stream()
        .map(Item::getMsg)
        .collect(Collectors.toList());
for(String str : list) {
    System.out.println(str);
}

流的创建

可以使用集合类的stream()或者parallelStream()方法创建流:

import java.util.stream.Stream;
Stream<String> s1 = Arrays.asList("a","b","c").stream();
Stream<String> s2 = Arrays.asList("a","b","c").parallelStream();

java.util.stream.Stream是一个interface, 各种管道中间操作的返回值都是它的实现类, 这允许我们方便地进行参数传递.

Stream的静态方法of()也可以用来创建流:

Stream.of(new int[]{1,2,3});

Arrays也提供了创建流的静态方法stream():

Arrays.stream(new int[]{1,2,3})

一些类也提供了创建流的方法:

IntStream.range(start, stop);
BufferedReader.lines();
Random.ints();

中间操作

流操作是惰性执行的, 中间操作会返回一个新的流对象, 当执行终点操作时才会真正进行计算.下面介绍流的中间操作. 除非传入的操作函数有副作用, 函数本身不会对数据源进行任何修改.

distinct

distinct保证数据源中的重复元素在结果中只出现一次, 它使用equals()方法判断两个元素是否相等.

List<String> list = Stream.of("a","b","c","b")
        .distinct()
        .collect(Collectors.toList());

filter

filter根据传入的断言函数对所有元素进行检查, 只有使断言函数返回真的元素才会出现在结果中. filter不会对数据源进行修改.

List<Integer> list = IntStream.range(1,10).boxed()
		.filter( i -> i % 2 == 0)
		.collect(Collectors.toList());

java.util.Objects提供了空元素过滤的工具:

List<MyItem> list = items.stream()
	.filter(Objects::nonNull)
	.collect(Collectors.toList());

map

map方法根据传入的mapper函数对元素进行一对一映射, 即数据源中的每一个元素都会在结果中被替换(映射)为mapper函数的返回值.

List<String> list = Stream.of('a','b','c')
        .map( s -> s.hashCode())
        .collect(Collectors.toList());

flatMap

与map不同flatMap进行多对一映射, 它要求若数据源的元素类型为R, 则mapper函数的返回值必须为Stream<R>.

flatMap会使用mapper函数将数据源中的元素一一映射为Stream对象, 然后把这些Stream拼装成一个流.因此我们可以使用flatMap进行合并列表之类的操作:

List<Integer> list = Stream.of(
		Arrays.asList(1),
		Arrays.asList(2, 3),
		Arrays.asList(4, 5, 6)
	)
	.flatMap(l -> l.stream())
	.collect(Collectors.toList());

peek

peek方法会对数据源中所有元素进行给定操作, 但在结果中仍然是数据源中的元素. 通常我们利用操作的副作用, 修改其它数据或进行输入输出.

List<String> list = Stream.of('a','b','c')
        .map(s -> System.out.println(s))
        .collect(Collectors.toList());

sorted

sorted方法用于对数据源进行排序:

List<Integer> list = Arrays.asList(1,2,3,4,5,6).stream()
		.sorted((a, b) -> a-b)
		.collect(Collectors.toList());

使用java.util.Comparator是更方便的方法, 默认进行升序排序:

class Item {
	int val;
	public Item(int val) { this.val = val; }
	public int getVal() { return val; }
}

List<Item> list = Stream.of(
		new Item(1),
		new Item(2), 
		new Item(3)
	)
	.sorted(Comparator.comparingInt(Item::getVal))
	.collect(Collectors.toList());

使用reversed()方法进行降序排序:

List<Item> list = Stream.of(
		new Item(1),
		new Item(2), 
		new Item(3)
	)
	.sorted(Comparator.comparingInt(Item::getVal).reversed())
	.collect(Collectors.toList());

limit

limit(int n)当流中元素数大于n时丢弃超出的元素, 否则不进行处理, 达到限制流长度的目的.

skip

skip(int)返回丢弃了前n个元素的流. 如果流中的元素小于或者等于n，则返回空的流

终点操作

reduce

reduce(accumulator)是最基本的终点操作之一, 操作函数accumulator接受两个参数x,y返回r.

reduce首先将数据源中的两个元素x1和x2传给accumulator得到r1, 然后将r1和x3传入得到r2. 如此进行直到处理完整个数据流.

reduce方法还可以接受一个参数代替x1作为起始值:

Integer sum = Stream.of(1,2,3,4,5).reduce(0, (x, y) -> x +y);
String concat = Stream.of("a", "b", "c", "d").reduce("", String::concat); 
double minValue = Stream.of(-1.5, 1.0, -3.0, -2.0)
	.reduce(Double.MAX_VALUE, Double::min); 

collect

collect是使用最广泛的终点操作, 也上文中多次出现:

List<String> list = Stream.of("a","b","c","b")
        .distinct()
        .collect(Collectors.toList())

toList()将流转换为List实例, 是最常见的用法, java.util.Collectors类中还有求和, 计算均值, 取最值, 字符串连接等多种收集方法.

forEach

forEach方法对流中所有元素执行给定操作, 没有返回值.

Stream.of(1,2,3,4,5).forEach(System.out::println);

其它

• count() 返回流中的元素数

• toArray(): 转换为数组

并发问题

除非显式地创建并行流, 否则默认创建的都是串行流.Collection.stream()为集合创建串行流,而Collection.parallelStream()创建并行流.

stream.parallel()方法可以将串行流转换成并行流,stream.sequential()方法将流转换成串行流.

流可以在非线程安全的集合上创建, 流操作不应该对非线程安全的数据源产生任何副作用, 否则将发生java.util.ConcurrentModificationException异常.

List<String> list = new ArrayList(Arrays.asList("a", "b"));
list = list.stream().forEach(s -> l.add("c")); // cause exception

对于线程安全的容器不会存在这个问题:

List<String> list = new CopyOnWriteArrayList<>(Arrays.asList("a", "b"));
list = list.stream().forEach(s -> l.add("c")); // safe operation

当然作者建议Stream操作不要对数据源进行任何修改. 当然, 修改其它数据或者输入输出是允许的:

list.stream().forEach(s -> {
	set.add(s); 
	System.out.println(s);
});

理想的管道操作应该是无状态且与访问顺序无关的. 无状态是指操作的结果只与输入有关, 下面即是一个有状态的操作示例:

State state = getState();
List<String> list = new ArrayList(Arrays.asList("a", "b"));
list = list.stream().map(s -> {
  if (state.isReady()) {
    return s;
  }
  else {
    return null;
  }
});

无状态的操作保证无论系统状态如何管道的行为不变, 与顺序无关则有利于进行并行计算.

函数式接口

函数式接口会将签名匹配的函数对象(lambda表达式或方法)视作接口的实现。

@FunctionalInterface
interface Greeter
{
    void hello(String message);
}

函数式接口中有且只有一个非抽象方法。

Greeter greeter = message -> System.out.println("Hello " + message);

这在 Java 8 之前通常使用匿名内部类实现的:

Greeter greeter = new Greeter() {
            @Override
            public void hello(String message) {
                System.out.println("Hello " + message);
            }
        };

Java 8 将已有的一些接口实现为函数式接口:

• java.lang.Runnable
• java.util.concurrent.Callable
• java.util.Comparator
• java.lang.reflect.InvocationHandler
• java.io.FileFilter
• java.nio.file.PathMatcher

java.util.function中定义了一些常用的函数式接口:

• Consumer: 接受参数无返回
  • Consumer<T> -> void accept(T t);
  • BiConsumer<T,U> -> void accept(T t, U u);
  • DoubleConsumer -> void accept(double value);
• Supplier: 不接受参数有返回
  • Supplier<T> -> T get();
  • DoubleSupplier -> double getAsDouble();
• Function: 接受参数并返回
  • Function<T, R> -> R apply(T t);
  • BiFunction<T, U, R> -> R apply(T t, U u);
  • DoubleFunction<R> -> R apply(double value);
  • DoubleToIntFunction -> int applyAsInt(double value);
  • BinaryOperator<T> extends BiFunction<T,T,T>
• Predicate: 接受参数返回boolean
  • Predicate<T> -> boolean test(T t);
  • BiPredicate<T, U> -> boolean test(T t, U u);
  • DoublePredicate -> boolean test(double value);

默认构造器可以作为supplier: Supplier<Item> supplier = Item::new;