如何避免写重复代码：善用抽象和组合

通常，为了简化我们对数据流的处理，我们可能会使用 Java8 中首次引入的 Stream 、或者是 Kotlin、Scala 等编程语言中提供的更加丰富的集合库，亦或者使用反应式流的相关三方库来简化工作。虽然这些类库已经提供了丰富的操作符，但是我们依然会工作中遇到其对某些场景未提供合适操作符的情况。比如：

1. 在直播场景下，需要对某些类型的消息进行缓冲和聚合，一段时间内的多个点赞合并为1个点赞，并且在处理了 N 个消息的时候进行整体发送，保障整体的扩散量级维持在一个平稳的水平。

2. 在 IOT 场景中，接收来自终端设备上报的数据，并返回当前的数据和前值，或者最近 3个值，从而计算其中的变化趋势。此时我们可能会使用反应式流库中提供的：zipWithNext、zipWithPrevious、zipWithPreviousAndNext，或者是 sliding 。

3. 在建立一个聊天室的时候，如果用户输入bye，则让用户断开连接，离开聊天室，那么这个时候我们可能会使用 takeWhile。

4. 假设我们有一组SQL，我们需要按照顺序执行，并合并他们的结果，并在处理完成后关闭对应的数据库连接，这时我们可能会用 mapWithResource, using（资源安全）。

5. 当处理文件、写入数据库等使用资源的时候，我们需要打开一个文件或获取一个数据库连接，将数据写入，然后在处理完成后关闭对应的资源，这时我们可能会使用 foldResource（资源安全）

6. 假设需要对数据进行分批，每 3个元素一批，进行打包，这个时候我们可能会使用 batch(3)

7. 假设我们需要将元素和每个元素的下标结合在一起，这个时候我们可能需要使用 zipWithIndex

8. 假设我们需要缓存元素，并在指定条件满足前一直缓存，我们可能需要 bufferUntil(predicate)、bufferWhile(predicate）●

9. 假设我们需要缓存元素，直到数据变更，把相同项合并在一起，我们可能需要 bufferUtilChanged

10. 假设我们需要对所有的元素进行去重，或者去掉连续的重复元素，我们可能会需要用到 distinct、distinctUntilChanged

11. 假设我们只需要返回前 N 个元素，我们可能需要使用 limit(N)、take(N), 或者按照条件 takeWhile、takeUntil假设我们需要跳过前N个元素，我们可能需要使用 skip(N)、drop(N), 或者按照条件 dropWhile、dropUntil。

12. 等等...



我们可以看到，上面这些操作符，每个都拥有具体的语义，虽然看起来只是一个简单的方法，但是如果需我们完全自主实现，定然也有不小的难度，比如 zipWithNext、zipWithPrevious、zipWithPreviousAndNext 在 Reactor-core 目前的发行版本中就没有直接提供，而和资源相关的， Reactor-core 中则只有一个 using 。

下面我们思考一下如何实现这些操作符吧~~

作为程序员，第一件事情，肯定就是 Ctrl + C ，第二件事就是 Ctrl + V， 第三件事就是 Commit & Push。然而，事情并没有这么简单。



难点有：

1. 反应式流操作符需要完整实现反应式流的规范、并通过默认的测试套件的验证。

2. 操作符需要尽可能的抽象和可组合。

3. 无论是单线程还是并发场景下都拥有正确的行为和语义、并有完整单元测试覆盖。

4. 操作符的实现需要尽可能的具备最高效的性能。



比如，以 zipWithIndex举例，在 Reactor-core 中有 FluxIndexFuseable（370行代码)和 FluxIndex (296行代码)两个实现。而且清晰的处理了各种情况。而其他操作符也有类似：release 3.4.23

1. FluxBuffer—— 575行代码

2. FluxBufferPredicate—— 464 行代码

3. FluxDistinct—— 609行代码

4. FluxDistinctFuseable—— 70行代码

5. FluxDistinctUntilChanged—— 337 行代码

6. FluxUsing—— 583 行代码

如果要实现一个zipWithNext 自定义操作符 ，应该也有接近的工作量。这样的工作强度，个人认为无论是在代码审查还是后期的维护都是一个大问题。



为此，我认为需要一个新的抽象，来对上面的这些操作进行进一步的抽象。然后再这个之上，通过使用和组合其他的操作，从而更简单的实现自定义操作符；

解法

1. 带有状态，且线程安全
2. 状态可变，且根据状态的不同，对输入应用不同的操作，产生不同的值
3. 可以提前结束、或者对不满足条件的值进行选择性丢弃
4. 有完整的生命周期
5. 在结束时可以根据内部状态而产生可选的值，而不会丢失内部状态
   

public <S, In, Out> statefulMap(    java.util.function.Supplier<S> create,    java.util.function.BiFunction<S, In, Pair<S, Out>> f,    java.util.function.Function<S, Optional<Out>> onComplete){...}

让我们看一下如何通过这个方法来实现 zipWithIndex吧：

▐  实现zipWithIndex (indexed)

Source.from(Arrays.asList("A", "B", "C", "D"))

    .statefulMap(        () -> 0L,        (index, element) -> Pair.create(index + 1, Pair.create(element, index)),        indexOnComplete -> Optional.empty())

    .runForeach(System.out::println, system);// prints// Pair(A,0)// Pair(B,1)// Pair(C,2)// Pair(D,3)

▐  实现 bufferUntilChanged

Source.from(Arrays.asList("A", "B", "B", "C", "C", "C", "D"))

    .statefulMap(        () -> (List<String>) new LinkedList<String>(),        (buffer, element) -> {          if (buffer.size() > 0 && (!buffer.get(0).equals(element))) {            return Pair.create(                new LinkedList<>(Collections.singletonList(element)),                Collections.unmodifiableList(buffer));          } else {            buffer.add(element);            return Pair.create(buffer, Collections.<String>emptyList());          }        },        Optional::ofNullable)    .filterNot(List::isEmpty)
    .runForeach(System.out::println, system);// prints// [A]// [B, B]// [C, C, C]// [D]

▐  实现 distinctUntilChanged

Source.from(Arrays.asList("A", "B", "B", "C", "C", "C", "D"))

    .statefulMap(        Optional::<String>empty,        (lastElement, element) -> {          if (lastElement.isPresent() && lastElement.get().equals(element)) {            return Pair.create(lastElement, Optional.<String>empty());          } else {            return Pair.create(Optional.of(element), Optional.of(element));          }        },        listOnComplete -> Optional.empty())    .via(Flow.flattenOptional())

    .runForeach(System.out::println, system);// prints// A// B// C// D

如果要实现聚合buffer呢？



▐  实现 buffer

Source.fromJavaStream(() -> IntStream.rangeClosed(1, 10))

    .statefulMap(        () -> new ArrayList<Integer>(3),        (list, element) -> {          list.add(element);          if (list.size() == 3) {            return Pair.create(new ArrayList<Integer>(3), Collections.unmodifiableList(list));          } else {            return Pair.create(list, Collections.<Integer>emptyList());          }        },        listOnComplete -> Optional.ofNullable(listOnComplete))    .filterNot(List::isEmpty)

    .runForeach(System.out::println, system);// printsList(1, 2, 3)List(4, 5, 6)List(7, 8, 9)List(10)

更复杂的例子：处理资源

1. 针对流中的每个元素都创建一个新的资源，使用这个资源，关闭这个资源。
2. 针对整个流创建一个资源，并在处理流中的每个元素时使用这个资源，并在流的生命周期结束后，关闭这个资源。

1. 资源的初始化和关闭需要支持并发安全；反应式流可以被多次物化，被多个下游订阅者订阅和处理，并且以任意的顺序进行取消订阅，需要在各种情况下（上游完成、下游取消、处理异常等）等情况下妥善的创建和销毁资源。
2. 在流生命周期的各个阶段安全地创建和销毁资源；比如：即使在创建资源或者销毁资源的时触发了异常，也不会对同一个资源关闭多次。
3. 支持异步从而提高资源使用的效率。
4. 感知流的生命周期，支持在关闭资源时提供可选的值给到下游以标识流的结束，比如处理文件时，使用一个特殊的标识符标识文件的结尾。

  public static <T, D> Mono<T> using(    Callable<? extends D> resourceSupplier,    Function<? super D, ? extends Mono<? extends T>> sourceSupplier,    Consumer<? super D> resourceCleanup) {...}

resourceSupplier 针对每个订阅者，创建一个资源

sourceSupplier 结合创建的资源，产生对应的元素

resourceCleanup 取消订阅或者流完成时，清理对应的资源

在 reactor-core 中，对应的底层实现为 MonoUsing 共 360 行代码，而要实现我们想要的逻辑，我们还需要和另一个流进行合并，即这里的 using 类似于 unfoldResource 。那么有没有可能使用更加简单的方案来进行实现呢？答案是肯定的，和前面的几个操作符一样，我们可以使用 statefulMap 来实现mapWithResource ，思维过程如下：

1. using / mapWithResource 的生命周期管理 和 statefulMap 的 create 和  onComplete 方法对应，针对资源，onComplete 方法可以被命名为更加贴切的 release / close / cleanUp。

2. 在流中使用的资源，我们可以认为是一个状态，只不过这个状态在流的整个生命周期中不再变化，一直是 create 方法中返回的 Resource 。

3. 在关闭资源时，我们可以通过返回一个 Optional<Out> 来返回一个可选的值。

4. 对并发资源的异步处理，则可以通过返回一个 CompletionStage<Out> 而非 Out 来实现，在 using 方法中，我们返回的是一个 Mono<T>

   

经过上面的思维过程，我们不难得出这个流上的方法的声明可以为：

public <R, In, Out> mapWithResource(  Supplier<? extends R> create,  BiFunction<? super R, ? super In, ? extends Out> function,  Function<? super R, ? extends Optional<? extends Out> close) {...}

resourceSupplier 针对每个订阅者/每次物化，创建一个资源

function 使用create 中创建的资源处理流中的每个元素

close 在流关闭的同时关闭资源，并再向下游提供一个可选的值

具体的的实现这里留空，感兴趣的小伙伴可以结合前面的例子进行实现。下面我们看一下如何使用这个 mapWithResource 方法，从而加深大家的理解。

▐  使用mapWithResource

Source.from(            Arrays.asList(                "SELECT * FROM shop ORDER BY article-0000 order by gmtModified desc limit 100;",                "SELECT * FROM shop ORDER BY article-0001 order by gmtModified desc limit 100;"))        .mapWithResource(            () -> dbDriver.create(url, userName, password),            (connection, query) -> db.doQuery(connection, query).toList(),            connection -> {              connection.close();              return Optional.empty();            })        .mapConcat(elems -> elems)        .runForeach(System.out::println, system);

在上面的例子中：我们有一组预先定义好的 SQL，分别从多个表中读取最新的 100 条数据，通过使用mapWithResource ，我们优雅地为每个流创建了 db 相关的连接，并进行对应的查询操作，并合并查询结果，在流处理完成后，关闭对应的资源。上面的代码通过复用我们前面编写的 mapWithResource 将复杂资源和生命周期管理进行了简化，作为对比，大家可以思考一下如果我们不使用已有抽象所需要付出的努力。

总结

团队介绍

大淘宝技术移动技术中台，推进淘宝、天猫等架构升级，致力于为淘系、整个集团提供基础核心能力、产品与解决方案：

1. 业务高可用的解决方案与核心能力（应用高可用：为业务提供自适应限流、隔离与熔断的柔性高可用解决方案，站点高可用：故障自愈、多机房与异地容灾与快速切流恢复）；

2. 新一代的业务研发模式FaaS（一站式函数研发Gaia平台）；

3. 下一代网络协议QUIC实现与落地；

4. 移动中间件（API网关MTop、域名调度AMDC、消息/发布订阅/推送、文件上传AUS、移动配置推送Orange 等等）。

服务端技术 | 技术质量 | 数据算法