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

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

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

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

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

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

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

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

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

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

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

• 假设我们只需要返回前N个元素，我们可能需要使用limit(N)、take(N)，或者按照条件takeWhile、takeUntil。

• 假设我们需要跳过前N个元素，我们可能需要使用skip(N)、drop(N)，或者按照条件dropWhile、dropUntil。

• ……
  



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

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){...}

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)

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]

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

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);
// prints
List(1, 2, 3)
List(4, 5, 6)
List(7, 8, 9)
List(10)

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

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) {...}

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);