SQL查询为什么还要用Stream的方法？

Posted 2021-05-22 风风风啊

在业务系统中，数据一般都从sql中查询（where,order by,limit,聚合函数等）为什么还要用Java8的stream的方法？

SQL（数据库）速度更快，如果数据量大而你有条件能用sql直接出结果，当然用sql呀。数据库有优化到极致的索引和缓存。

但是目前流行的微服务架构上，不同的服务用的是独立的数据库，那一般来说你就没法直接简单用sql来搞了，通常你是先找服务A要了一堆数据，然后在你程序里把这堆数据处理成下一步的请求发给服务B和服务C，这两个服务分别返回各自的数据，然后你的程序把这两堆数据合并成一个结果返回。

又或者，数据库中存储的格式不是你要返回的格式，你需要做一些复杂的转换再返回，用sql的话你得用储存过程，而大部分现代的系统一般不提倡使用储存过程，所以你就得在程序里做。

当然这些东西你完全可以用循环或者iterator来做，java8之前大家一直都是这么玩的。但是在某些场合，Stream做起来更方便直观。所以Stream是用来代替循环和iterator的，跟sql并没有什么竞争关系。能用相对简单的sql出结果的，那就用sql。速度快，调试又方便。

如果不考虑多核方面的好处的话，Stream在日常编程里最大好处就是抽象迭代逻辑。Java8之前这个任务是用iterator来负责的，但iterator太重，要自己实现一个独立的类不说，hasNext和next的逻辑通常就十分反人类，一般人都不想去搞。另外iterator方案本质上其实治标不治本，你用它来封装迭代逻辑，但用的时候iterator本身又需要用一个循环来驱动，总是留点尾巴，在使用iterator过程中有一些简单的额外处理逻辑你就开始纠结究竟是写在驱动的那个循环里还是再套一层iterator。大部分人干脆就不去抽象，一堆循环加一堆状态变量把业务逻辑和迭代逻辑混在一起对付过去就算了。

Stream在某些场景下大大简化了抽象迭代逻辑的工作。这种抽象方案在clojure或者haskell这类没有循环语法的语言中很常用，叫“惰性序列”。Java 8之后利用Stream也可以用起来了。

举个具体例子吧，假如你在做个任务跟踪软件，你可以定义一些每周重复任务，每个任务可以设置开始日期，结束日期和一个DayOfWeek（星期几）列表。表示在这段时间内，每逢指定的DayOfWeek，就执行该任务。结束日期是可选的，如果你没设置结束日期，那就代表这个任务一直重复。任务可以分组，同一组里可能有多个任务。现在我要按时间顺序显示一个组里所有任务的执行计划。也就是需要把上面由“时间区间+DayOfWeek”所表示的任务执行时间，展开成具体的执行日期，并且按照时间顺序把一个组内所有任务的执行日期合并成一条时间线。而且，显示执行计划的界面有两个，一个是无限滚动条的分页界面，请求提供一个页面大小pageSize，你每次返回pageSize条任务计划，一个是日历界面，请求提供一个日期区间，你返回这个日期区间内的所有任务记录。

你先想想如果用循环，你会怎么做，业务逻辑和循环代码是不是会混在一起。

下面来看看用Stream可以怎么做。（我就随手想了个例子来举例，未必是最优方案，欢迎友善讨论）

首先处理“日期范围+星期”转化成具体日期的问题。

在没有Stream之前，实现这个需求一般是用队列来做，也就是先把第一周符合条件的日期放入队列（但期间你要判断是否已经超过了endDate），然后再在队列中取出一个日期进行业务处理，并且计算一周后的日期重新加入队列。直到到达处理边界（到达endDate或pageSize）。注意这种做法使用循环的话，业务逻辑一般放在循环内部，也就是迭代逻辑分布在业务两端，根本难以区分。也可以用iterator来做，但hasNext和next方法的实现需要引入一些状态变量，比较复杂，一般在第一次遇到时不会优先采用iterator方案，只有在后续开发发现同类转换很多时才考虑重构。

可以创建一个Stream来获取某个任务的执行日期。

/**
 * 给出时间区间和一个指定星期几的集合，返回一个包含符合条件的日期的Steam。日期按先后顺序排序。
 * 如果省略结束日期（endDate为null），则返回一个无穷Stream，客户代码根据需要自行将其截断（使
 * 用takeWhile, limit等方法）。
 */
public static Stream<LocalDate> getDatesInRange(@Nonnull LocalDate startDate, @Nullable LocalDate endDate, @Nullable Set<DayOfWeek> dayOfWeeks) {
        Objects.requireNonNull(startDate, "Must provide a start date");
        // 创建第一周的日期列表
        final LocalDate oneWeekLater = startDate.plusWeeks(1);
        final List<LocalDate> datesInAWeek = Stream.iterate(startDate, d -> d.isBefore(oneWeekLater), d -> d.plusDays(1))
                                                   .filter(d -> dayOfWeeks == null || dayOfWeeks.contains(d.getDayOfWeek()))
                                                   .collect(Collectors.toList());
        // 创建一个无穷Stream，每次通过前一周的日期列表计算下一周的日期列表（表中每个日期后推一周）
        // 然后用flatMap铺平为日期的Stream。
        final Stream<LocalDate> dateStream = Stream.iterate(
            datesInAWeek,
            dates -> dates.stream().map(d -> d.plusWeeks(1)).collect(Collectors.toList())
        ).flatMap(Collection::stream);
        // 如果指定了结束日期，则选取到结束日期为止
        return endDate != null
            ? dateStream.takeWhile(d -> !d.isAfter(endDate))
            : dateStream;
}
复制代码

如果习惯了面向Stream编程，上面的代码其实逻辑是很清楚的。首先，创建一个List包含第一周符合条件的日期。那么第二周的日期列表就是第一周List中的每个日期后推一周，如此类推。因此可以用Stream.iterate创建一个“每周日期列表”的Stream。Stream里的每个元素是每周符合条件的日期所组成的List。然后用Stream上的flatMap方法把这些列表摊平，那么就得到了一个日期的Stream。

这个做法有不少好处：

1. 你不用管创建下一个“每周日期列表”的时机，Stream消费了前一周的列表就会自己去创建下一周的列表。

2. 你不用管怎么把每周日期列表里的日期提取出来变成一个日期Stream。

3. 可选的endDate处理很方便，有就用takeWhile截断，没有就把整个无穷Stream直接返回。

4. 整个方法中三个部分分隔得非常清晰，两个赋值一个return，刚好完成三个任务。没有复杂的循环分支结构。赋值全是final，没有可变变量，下一步仅仅依赖于上一步创建的那个变量，没有贯穿整个方法的公共状态变量。

5. 整个方法只管创建这个日期Stream，完全不涉及任何具体业务逻辑。客户代码拿到这个Stream后怎么用都跟这个方法无关了。由于不涉及业务逻辑，这个方法可以用在任何需要“日期范围+星期“转换成日期的场景。由于写法相对简单，第一次遇见时觉得这里可以抽象一下，随手找个现有的Util类写个静态方法就好了。

好了，现在对于每一个任务，你都能拿到一个日期Stream。对于一个组内的多个任务，你就能拿到一个日期Stream的列表（List<Stream>)。下一步，需要把这些列表里的日期按顺序取出来，注意由于可以存在无穷Stream，你不能直接flatMap之后sorted。即使不是无穷Stream，这样做效率也十分低，因为sorted需要先把每个stream的元素都提取出来再统一排序。

这里其实又可以抽象出来一个跟业务无关的逻辑：合并多个已经排序的Stream，形成一个新的在多个Stream中取元素的Stream。自然，如果传入的Stream中有无穷Stream，那么这个合并后的Stream也是无穷的。

具体做法是，在传入的多个Stream里各取第一个元素，放进一个优先队列里（PriorityQueue，它的poll方法总是返回队列中最小的那个元素），然后在优先队列中取第一个元素，作为结果Stream的当前元素。如果这个元素所在的Stream还未耗尽，则再在这个Stream里取下一个元素，放入优先队列。（这跟前面提到的用队列来求日期列表方法思路大致相同）直到优先队列为空。

这里有个问题，Stream的接口本身不支持仅取一个元素进行一些操作后接着取后续元素。如果你用firstFirst方法，第二次再取调用这个stream上的任何方法就会抛异常，说steam已经被使用过了。因此只能把stream转化成iterator再用，幸好Stream上就有.iterator方法可以直接获取对应的iterator（可见Stream和Iterator本质上是一个东西）。

另一个需要考虑的地方是，如果优先队列里直接放元素，在取出元素后没有办法把元素和它来源的Iterator联系起来。所以在优先队列里实际放的应该是一个同时包含元素和来源Iterator的数据结构，在下面的例子我使用了apache common里的Pair。

/**
  * 接收一个Stream集合和一个Comparator，假定所有传入Stream会按照Comparator所定义的顺序供给元素。
  * 返回一个Stream来按顺序获取所有Stream集合中的元素，排序规则由Comparator定义。Stream中的null值
  * 将会被丢弃。
 */
private static  <T> Stream<T> mergeSortedStream(Collection<Stream<T>> streams, Comparator<T> comparator) {
    final PriorityQueue<Pair<T, Iterator<T>>> priorityQueue = new PriorityQueue<>(
        (p1, p2) -> comparator.compare(p1.getKey(), p2.getKey())
    );
    // 初始化优先队列
    priorityQueue.addAll(
        streams.stream().map(s -> {
            final Iterator<T> itr = s.iterator();
            if (itr.hasNext()) {
                return Pair.of(itr.next(), itr);
            } else {
                return null;
            }
        })
        .filter(Objects::nonNull)
        .collect(toList())
    );
    // 创建Stream并返回
    return Stream.generate(() -> {
        final Pair<T, Iterator<T>> head = priorityQueue.poll();
        if (head != null) {
            final T item = head.getKey();
            final Iterator<T> itr = head.getValue();
            if (itr.hasNext()) {
                // 跳过null值，如果null值进入优先队列会影响排序逻辑
                T next;
                do {
                    next = itr.next();
                } while (next == null && itr.hasNext());
                if (next != null) {
                    priorityQueue.add(Pair.of(next, itr));
                }
            }
            return item;
        } else {
            return null;
        }
    }).takeWhile(Objects::nonNull);
}
         
复制代码

代码有点长，由于需要使用iterator所以有一堆的if和循环，基本思路如前所述。因为是通用逻辑，所以接收一个Comparator允许客户代码定义排序规则。

可以看出，这个静态方法同样不涉及任何业务逻辑。以后所有已排序Stream的归并都可以用它。

最后，我们只要

Stream<LocalDate> taskCategoryDateStream = mergeSortedStream(taskDateStreams, Comparator.comparing(Function.identity()));
复制代码

就行了。拿到的Stream可能是无穷的，我们根据业务需要，如果要按pageSize提取那就是调limit，如果要按endDate提取那就是takeWhile（由于可能是无穷Stream不要用filter）。接着forEach也行，reduce也行，collect也行，随便弄，业务逻辑这时才进场。相对于循环和迭代器，Stream方案（惰性序列）另一个好处是Stream上的方法非常丰富，拿到Stream后你想forEach, groupingBy，toMap，findAny都很方便，不用写一大堆嵌套循环而其实每个循环只做一点点东西。

以上就是一个面向Stream编程的例子，可以看出，跟sql完全是不同维度的东西。

首发地址：SQL查询为什么还要用Stream的方法？

关注技术号：公众号：码农架构

关键词：站长资讯中心，站长资讯，站长新闻