我想對 Flux 執(zhí)行移動窗口計(jì)算并生成包含計(jì)算值的 Flux，但我不知道如何完成此操作。作為一個(gè)簡化的例子，假設(shè)我有一個(gè)整數(shù)通量，我想用這個(gè)通量中每 3 個(gè)連續(xù)整數(shù)的總和生成一個(gè)新的通量。為了顯示：第一個(gè)通量包含從 1 到 8 的整數(shù)：{1, 2, 3, 4, 5, 6, 7, 8}結(jié)果通量應(yīng)包含總和：{1+2+3、2+3+4、3+4+5、4+5+6、5+6+7、6+7+8}我可以輕松地生成第一個(gè) Flux 并導(dǎo)出包含連續(xù) 3 個(gè)值的通量通量，如下所示：Flux<Integer> f1 = Flux.range(1,8);
Flux<Flux<Integer>> f2 = f1.window(3,1);我也可以訂閱 () 到 f2 并計(jì)算總和，但我不知道如何同時(shí)將這些總和發(fā)布為新的 Flux。我錯過了一些簡單的事情，還是這種事情真的很難做？
1 回答
HUWWW
TA貢獻(xiàn)1874條經(jīng)驗(yàn) 獲得超12個(gè)贊
您可以使用.reduce(Integer::sum)內(nèi)部通量來執(zhí)行窗口中元素的總和，并.flatMap使用外部通量將這些總和合并回單個(gè)流。
請注意，由于.window調(diào)用 with maxSize < skip，因此尾隨窗口中的項(xiàng)目將小于最大大小。
Flux<Integer> sums = Flux.range(1, 8) // Flux<Integer>
.window(3, 1) // Flux<Flux<Integer>>
.flatMap(window -> window.reduce(Integer::sum)); // Flux<Integer>
StepVerifier.create(sums)
.expectNext(6) // 1+2+3
.expectNext(9) // 2+3+4
.expectNext(12) // 3+4+5
.expectNext(15) // 4+5+6
.expectNext(18) // 5+6+7
.expectNext(21) // 6+7+8
.expectNext(15) // 7+8
.expectNext(8) // 8
.verifyComplete();
添加回答
舉報(bào)
0/150
提交
取消