simple-stream: Java8 Stream的简易实现

前言

Java8新增的Stream API是一个强大的特性，它可以简化集合中的常用操作，包括过滤、分组等。下面就来实现一个简易版的Stream。

从表面上看，流似乎和列表很接近，但实际上它们有着本质的区别。

列表是多个元素的容器，当列表被创建出来时，它里面的每个元素也已经被创建出来了。
流是一种计算结构，它封装了内部元素如何产生的计算过程，但是并没有包含实际的元素数据。换句话说，当一个流被创建出来时，它内部的元素并没有被创建，但是我们可以通过调用流的方法来按顺序生成每个元素。

所以，流具有惰性计算的特性，它可以表示普通列表无法表示的一些结构，如无限流。

流的定义

流的定义看起来很像链表，一个流由两部分组成：第一个元素（first）和剩余元素组成的流（remain）。定义如下：

public interface Stream<T> {
    /**
     * 流中第一个元素
     */
    T first();

    /**
     * 剩余元素组成的流
     */
    Stream<T> remain();

    /**
     * 创建流
     * @param firstSupplier 第一个元素的工厂
     * @param remainSupplier 剩余元素组成的流的工厂
     * @param <T> 元素类型
     * @return 流
     */
    static <T> Stream<T> create(Supplier<T> firstSupplier, Supplier<Stream<T>> remainSupplier) {
        return new Stream<>() {
            @Override
            public T first() {
                return firstSupplier.get();
            }

            @Override
            public Stream<T> remain() {
                return remainSupplier.get();
            }
        };
    }
}

这种递归的定义非常有利于使用递归算法来操作流。下面可以看到，流的大多数相关操作都是用递归算法实现的。

假设我们已经有了一个流，那么如何获取流中的元素呢？首先调用first来获取第一个元素，然后调用remain().first()来获取第二个元素，依此类推：

Stream<Integer> stream = ...
Integer first  = stream.first(); // 第一个元素
Integer second = stream.remain().first(); // 第二个元素
Integer third  = stream.remain().remain().first(); // 第三个元素

// 依此类推...

当然，我们不会用这种方法来访问流中的元素。具体如何访问，请继续往下看。

空流

空流是最简单的流，无法从空流中获取任何元素。空流也标志着一个流的结束。下面是空流的实现：

Stream<?> EMPTY = create(
        () -> {throw new IllegalStateException("当前流已结束");},
        () -> {throw new IllegalStateException("当前流已结束");}
);


/**
 * 获取空流
 */
@SuppressWarnings("unchecked")
static <T> Stream<T> empty() {
    return (Stream<T>) EMPTY;
}

/**
 * 判断当前流是否结束
 */
default boolean end() {
    return this == EMPTY;
}

有限流的生成

有限流可以通过多种方式生成，包括从数组生成、从迭代器生成、从集合生成。

从数组生成流

/**
 * 从数组生成流
 * @param arr 数组
 * @param <T> 元素类型
 * @return 流
 */
@SafeVarargs
static <T> Stream<T> of(T... arr) {
    return fromArray(0, arr);
}

/**
 * 从数组和起始索引生成流
 * @param startIndex 起始索引
 * @param arr 数组
 * @param <T> 元素类型
 * @return 流
 */
static <T> Stream<T> fromArray(int startIndex, T[] arr) {
    return startIndex == arr.length
            ? empty()
            : create(() -> arr[startIndex], () -> fromArray(startIndex + 1, arr));
}

从迭代器生成流

/**
 * 从迭代器生成流
 * @param iterator 迭代器
 * @param <T> 元素类型
 * @return 流
 */
static <T> Stream<T> fromIterator(Iterator<T> iterator) {
    return iterator.hasNext()
            ? create(iterator::next, () -> fromIterator(iterator))
            : empty();
}

从集合生成流

/**
 * 从集合生成流
 * @param collection 集合
 * @param <T> 元素类型
 * @return 流
 */
static <T> Stream<T> fromCollection(Collection<T> collection) {
    return fromIterator(collection.iterator());
}

示例

Stream<Integer> s1 = Stream.of(1, 2, 3); // 从数组生成
Stream<Integer> s2 = Stream.fromIterator(List.of(1, 2, 3).iterator()); // 从迭代器生成
Stream<Integer> s3 = Stream.fromCollection(Set.of(1, 2, 3)); // 从集合生成

无限流的生成

无限流意味着流中的元素个数没有限制，也就是永远都不会结束，所以end方法调用永远为false。有以下两种方法生成无限流。

从工厂方法生成流

/**
 * 从工厂方法生成流
 * @param supplier 生成流中元素的工厂方法
 * @param <T> 元素类型
 * @return 流
 */
static <T> Stream<T> fromSupplier(Supplier<T> supplier) {
    return create(supplier, () -> fromSupplier(supplier));
}

从生成器生成流

/**
 * 迭代生成流
 * @param initial 初始值
 * @param generator 生成器
 * @param <T> 元素类型
 * @return 流
 */
static <T> Stream<T> fromGenerator(T initial, UnaryOperator<T> generator) {
    return create(() -> initial, () -> generate(generator.apply(initial), generator));
}

示例

Stream<Integer> s1 = Stream.fromSupplier(() -> 1); // 无限个1组成的流
Stream<Integer> s2 = Stream.fromGenerator(1, n -> n + 1); // 全体自然数组成的流

遍历流中的元素

知道了如何创建流，那么如何遍历或输出流中的元素呢？可以实现下面的forEach方法：

/**
 * 遍历流中所有元素
 * @param consumer 遍历操作
 */
default void forEach(Consumer<T> consumer) {
    Stream<T> s = this;
    while (!s.end()) {
        consumer.accept(s.first());
        s = s.remain();
    }
}

然后就可以像下面这样输出流中的元素：

Stream<Integer> s = Stream.of(1, 2, 3, 4, 5);
s.forEach(System.out::println); // 输出1 2 3 4 5

流的截断和偏移

上面的forEach方法只适用于有限流，如果在无限流上调用forEach方法，会导致死循环。所以，我们需要对无限流进行截取操作，这样就能做到遍历无限流的一部分。

/**
 * 截取流中前n个元素
 * @param n 要截取的元素个数
 * @return 流
 */
default Stream<T> limit(int n) {
    return n <= 0 || end()
            ? empty()
            : create(this::first, () -> remain().limit(n - 1));
}

/**
 * 跳过流中的元素
 * @param n 跳过的个数
 * @return 流
 */
default Stream<T> skip(int n) {
    return end() || n <= 0
            ? this
            : remain().skip(n - 1);
}

limit用于提取流的前n个元素，skip用于忽略流的前n个元素，有了这两个方法，我们就能随心所欲地截取任何流中的任意一段。

流的变换操作

熟悉Java8 Stream API的读者一定用过map和filter这两个常用的流操作，下面我们就来实现它们。

map

map用于对流中的所有元素进行转换操作。

/**
 * 映射流中的元素
 * @param mapper 映射器
 * @param <U> 映射后的元素类型
 * @return 流
*/
default <U> Stream<U> map(Function<T, U> mapper) {
    return end()
            ? empty()
            : create(() -> mapper.apply(first()), () -> remain().map(mapper));
}

filter

filter用于过滤流中的元素。

/**
 * 过滤流中的元素
 * @param predicate 断言
 * @return 流
 */
default Stream<T> filter(Predicate<T> predicate) {
    if (end()) {
        return empty();
    }
    T e = first();
    if (predicate.test(e)) {
        return Stream.create(() -> e, () -> remain().filter(predicate));
    } else {
        return remain().filter(predicate);
    }
}

示例

Stream<String> s = Stream.of(1, 2, 3, 4, 5, 6)
        .filter(n -> n % 2 == 0) // 2, 4, 6
        .map(n -> "hello " + n); // hello 2, hello 4, hello 6

流的聚合操作

有时候我们像将整个流聚合成某种数据结构，如列表、集合等，这就需要用到流的聚合操作。

collect

对流进行自定义聚合操作。

/**
 * 流的聚合操作
 * @param initial 初始值
 * @param accumulator 聚合操作
 * @param <U> 聚合后的类型
 * @return 流
 */
default <U> U collect(U initial, BiFunction<U, T, U> accumulator) {
    U result = initial;
    Stream<T> s = this;
    while (!s.end()) {
        result = accumulator.apply(result, s.first());
        s = s.remain();
    }
    return result;
}

toList

将流转换成列表。

/**
 * 将流转换成列表
 * @return 列表
 */
default List<T> toList() {
    return collect(new ArrayList<>(), (list, e) -> {
        list.add(e);
        return list;
    });
}

toSet

将流转换成集合。

/**
 * 将流转换成集合
 * @return 集合
 */
default Set<T> toSet() {
    return collect(new HashSet<>(), (set, e) -> {
        set.add(e);
        return set;
    });
}

toMap

将流转换成Map。

/**
 * 将流转换成map
 * @param keyGenerator key生成器
 * @param valueGenerator value生成器
 * @param <K> key的类型
 * @param <V> value的类型
 * @return map
 */
default <K, V> Map<K, V> toMap(Function<T, K> keyGenerator, Function<T, V> valueGenerator) {
    return collect(new HashMap<>(), (map, e) -> {
        map.put(keyGenerator.apply(e), valueGenerator.apply(e));
        return map;
    });
}

count

对流中的元素进行计数。

/**
 * 获取流中元素个数
 * @return 元素个数
 */
default int count() {
    return collect(0, (cnt, e) -> cnt + 1);
}

流的高级操作

下面是流的一些高级操作。

concat

concat用于将两个流首尾连接在一起。

/**
 * 首尾连接两个流
 * s1[0] -> s1[1] -> s2[2] ->  ... -> s2[0] -> s2[1] -> s2[2] -> ...
 * @param s1 s1
 * @param s2 s2
 * @param <T> 元素类型
 * @return 流
 */
static <T> Stream<T> concat(Stream<T> s1, Stream<T> s2) {
    return s1.end()
            ? s2
            : create(s1::first, () -> concat(s1.remain(), s2));
}

/**
 * 首尾连接两个流
 * @param s 要连接的流
 * @return 流
 */
default Stream<T> concat(Stream<T> s) {
    return concat(this, s);
}

示例：

Stream<Integer> s1 = Stream.of(1, 2, 3, 4);
Stream<Integer> s2 = Stream.of(5, 6, 7);
Stream<Integer> s = s1.concat(s2); // 1, 2, 3, 4, 5, 6, 7

interleave

interleave用于将两个流交错连接在一起。

/**
 * 交错连接两个流
 * s1[0] -> s2[0] -> s1[1] -> s2[1] -> ...
 * @param s1 s1
 * @param s2 s2
 * @param <T> 元素类型
 * @return 流
 */
static <T> Stream<T> interleave(Stream<T> s1, Stream<T> s2) {
    return s1.end()
            ? s2
            : create(s1::first, () -> interleave(s2, s1.remain()));
}

/**
 * 交错连接两个流
 * @param s 要连接的流
 * @return 流
 */
default Stream<T> interleave(Stream<T> s) {
    return interleave(this, s);
}

示例：

Stream<Integer> s1 = Stream.of(1, 3, 5, 7);
Stream<Integer> s2 = Stream.of(2, 4, 6);
Stream<Integer> s = s1.interleave(s2); // 1, 2, 3, 4, 5, 6, 7

flatMap

flatMap用于将流中的每个元素都映射成一个流，然后将所有流连接起来。

/**
 * 扁平化流
 * @param mapper 元素到流的映射器
 * @param <U> 扁平化后的元素类型
 * @return 流
 */
default <U> Stream<U> flatMap(Function<T, Stream<U>> mapper) {
    return collect(empty(), (s, e) -> s.concat(mapper.apply(e)));
}

示例：

Stream<Integer> s = Stream.of(10, 20)
        .flatMap(n -> Stream.of(n + 1, n + 2, n + 3)); // 11, 12, 13, 21, 22, 23

后记

完整代码：https://github.com/byx2000/simple-stream

byx / simple-stream

前言

流的定义

空流

有限流的生成

从数组生成流

从迭代器生成流

从集合生成流

示例

无限流的生成

从工厂方法生成流

从生成器生成流

示例

遍历流中的元素

流的截断和偏移

流的变换操作

map

filter

示例

流的聚合操作

collect

toList

toSet

toMap

count

流的高级操作

concat

interleave

flatMap

后记

简介

发行版

贡献者

近期动态

byx / simple-stream .gitee-modal { width: 500px !important; }

前言

流的定义

空流

有限流的生成

从数组生成流

从迭代器生成流

从集合生成流

示例

无限流的生成

从工厂方法生成流

从生成器生成流

示例

遍历流中的元素

流的截断和偏移

流的变换操作

map

filter

示例

流的聚合操作

collect

toList

toSet

toMap

count

流的高级操作

concat

interleave

flatMap

后记

简介

发行版

贡献者

近期动态

搜索帮助

byx / simple-stream