跳到主要内容

三种批处理方式

Doubao 中英对照 Three Sorts of Batching

目录

当您有大量元素、想把它们划分成批时，Reactor 中有三大类方案：分组（grouping）、窗口（windowing）和缓冲（buffering）。这三者在概念上很接近，因为它们都是把一个 Flux<T> 重新聚合成一个聚合体。分组和窗口会产生 Flux<Flux<T>>，而缓冲则聚合为 Collection<T>。

1. 用 Flux<GroupedFlux<T>> 分组​

分组是指把源 Flux<T> 拆分成多个批，每一批对应一个 key。

对应的操作符是 groupBy。

每个分组都表示为一个 GroupedFlux<T>，您可以调用它的 key() 方法取出对应的 key。

分组的内容并不要求连续。一旦某个源元素产生了新的 key，就会为该 key 打开一个分组，此后匹配该 key 的元素都会进入这个分组（同时可能有多个分组处于打开状态）。

这意味着分组：

  1. 始终互不相交（一个源元素属于且仅属于一个分组）。

  2. 可以包含来自原始序列中不同位置的元素。

  3. 永不为空。

下面这个示例按数值是偶数还是奇数进行分组：

StepVerifier.create(
	Flux.just(1, 3, 5, 2, 4, 6, 11, 12, 13)
		.groupBy(i -> i % 2 == 0 ? "even" : "odd")
		.concatMap(g -> g.defaultIfEmpty(-1) //if empty groups, show them
				.map(String::valueOf) //map to string
				.startWith(g.key())) //start with the group's key
	)
	.expectNext("odd", "1", "3", "5", "11", "13")
	.expectNext("even", "2", "4", "6", "12")
	.verifyComplete();
注意

分组最适合分组数量中等偏少的场景。这些分组还必须被消费（例如通过 flatMap），groupBy 才会继续从上游拉取数据并供给更多分组。有时这两个约束会叠加在一起并导致挂起，例如当基数很高、而消费这些分组的 flatMap 并发度又太低时。

为了更好地理解把 groupBy 与不合适的操作符组合时挂起的风险，我们来看一个示例。

下面这段代码按字符串的第一个字符进行分组：

public static Flux<String> createGroupedFlux() {
        List<String> data = List.of("alpha", "air", "aim", "beta", "cat", "ball", "apple", "bat", "dog", "ace");
        return Flux.fromIterable(data)
                .groupBy(d -> d.charAt(0), 5)
                .concatMap(g -> g.map(String::valueOf)
                        .startWith(String.valueOf(g.key()))
                        .map(o -> {
                            System.out.println(o);
                            return o;
                        })
                );
    }

    @Test
    public void testGroupBy() {
        StepVerifier.create(createGroupedFlux())
                .expectNext("a", "alpha", "air", "aim", "apple", "ace")
                .expectNext("b", "beta", "ball", "bat")
                .expectNext("c", "cat", "d", "dog")
                .verifyComplete();
    }

在上面的例子中：

  • 分组的基数为 4（分组 key 为 "a"、"b"、"c" 和 "d"）。

  • concatMap 的并发度为 1。

  • groupBy 的缓冲大小为 5（因为我们把 prefetch 设为了 5；默认值是 256）。

a alpha air aim apple

在打印出这些元素之后，测试就卡住了。我们来分析一下发生了什么。

  1. 最初，groupBy 请求 5 个元素。它们进入一个缓冲："alpha", "air", "aim", "beta", "cat"。

  2. concatMap 的并发度为 1。因此，只有 key 为 "a" 的分组被订阅了。在最初的这些元素中，"alpha", "air", "aim" 被 concatMap 消费，而 "beta", "cat" 留在缓冲中。

  3. 接着，groupBy 再请求 3 个元素（缓冲中已有 2 个元素）。此时缓冲里是 "beta", "cat", "ball", "apple", "bat"。其中 "apple" 被消费，其余留在缓冲中。

  4. 接着，groupBy 再请求 1 个元素（缓冲中已占了 4 个位置）。缓冲中的元素是 "beta", "cat", "ball", "bat","dog"。

  5. 此时缓冲中已经没有任何属于 "a" 分组的元素，因此 concatMap 不再有任何消费动作，第一个 Flux 也就迟迟不能完成。groupBy 由于缓冲已满，无法再向 publisher 请求更多数据。publisher 遇到背压，无法再发布剩余元素。这就导致了死锁。

在同一个示例中，如果数据的顺序稍有不同，例如：

"alpha", "air", "aim", "beta", "cat", "ball", "apple", "dog", "ace", "bat"

那么同一个测试就能顺利通过：同样的 concatMap 就能收到完成信号、完成一个分组，然后订阅下一个分组，依此类推。因此，当发布的数据模式是任意的时候，如果消费模式与 groupBy 缓冲的容量不匹配，groupBy 很容易遇到死锁。

2. 用 Flux<Flux<T>> 开窗​

开窗是指按大小、时间、界定边界的谓词或界定边界的 Publisher 等条件，把源 Flux<T> 切分成一个个窗口。

对应的操作符有 window、windowTimeout、windowUntil、windowWhile 和 windowWhen。

与 groupBy 会根据到来的 key 随机重叠不同，窗口（大多数时候）是顺序打开的。

不过有些变体仍然可以重叠。例如，在 window(int maxSize, int skip) 中，maxSize 参数表示攒够多少个元素后关闭一个窗口，而 skip 参数表示源中再经过多少个元素后打开一个新窗口。因此，如果 maxSize > skip，新窗口会在前一个窗口关闭之前就打开，两个窗口就会重叠。

下面这个示例展示了相互重叠的窗口：

StepVerifier.create(
	Flux.range(1, 10)
		.window(5, 3) //overlapping windows
		.concatMap(g -> g.defaultIfEmpty(-1)) //show empty windows as -1
	)
		.expectNext(1, 2, 3, 4, 5)
		.expectNext(4, 5, 6, 7, 8)
		.expectNext(7, 8, 9, 10)
		.expectNext(10)
		.verifyComplete();
备注

在相反的配置下（maxSize < skip），源中的一部分元素会被丢弃，不属于任何窗口。

对于通过 windowUntil 和 windowWhile 基于谓词开窗的情况，如果后续的源元素不满足谓词，也可能产生空窗口，如下例所示：

StepVerifier.create(
	Flux.just(1, 3, 5, 2, 4, 6, 11, 12, 13)
		.windowWhile(i -> i % 2 == 0)
		.concatMap(g -> g.defaultIfEmpty(-1))
	)
		.expectNext(-1, -1, -1) //respectively triggered by odd 1 3 5
		.expectNext(2, 4, 6) // triggered by 11
		.expectNext(12) // triggered by 13
		// however, no empty completion window is emitted (would contain extra matching elements)
		.verifyComplete();

3. 用 Flux<List<T>> 缓冲​

缓冲与开窗类似，区别在于：它发出的不是窗口（每个窗口本身又是一个 Flux<T>），而是缓冲（即 Collection<T>——默认是 List<T>）。

缓冲对应的操作符与开窗一一对应：buffer、bufferTimeout、bufferUntil、bufferWhile 和 bufferWhen。

每当对应的开窗操作符打开一个窗口时，缓冲操作符就创建一个新集合并开始往里添加元素；每当窗口关闭时，缓冲操作符就发出这个集合。

缓冲同样可能丢弃源元素，或产生相互重叠的缓冲，如下例所示：

StepVerifier.create(
	Flux.range(1, 10)
		.buffer(5, 3) //overlapping buffers
	)
		.expectNext(Arrays.asList(1, 2, 3, 4, 5))
		.expectNext(Arrays.asList(4, 5, 6, 7, 8))
		.expectNext(Arrays.asList(7, 8, 9, 10))
		.expectNext(Collections.singletonList(10))
		.verifyComplete();

与开窗不同，bufferUntil 和 bufferWhile 不会发出空缓冲，如下例所示：

StepVerifier.create(
	Flux.just(1, 3, 5, 2, 4, 6, 11, 12, 13)
		.bufferWhile(i -> i % 2 == 0)
	)
	.expectNext(Arrays.asList(2, 4, 6)) // triggered by 11
	.expectNext(Collections.singletonList(12)) // triggered by 13
	.verifyComplete();