在 Reactor 3 中将 Flux 拆分为多个 Flux 的最有效方法 [英] The most efficient way to split a Flux to multiple Fluxes in Reactor 3

问题描述

在 Reactor 3 中，通过模式匹配将异构通量拆分为多个通量的最有效方法是什么?(而且后续对每个flux的操作可能会有很大的不同)

In Reactor 3, what's the most efficient way to split a heterogeneous flux to multiple fluxes by pattern matching? (And subsequent operations on each flux may be very different)

例如

Source Flux: a->b->c->a->b->c
 ||
 vv
A Flux: a->a->a
B Flux: b->b->b
C Flux: c->c->c

我是响应式编程的新手，我想出的唯一解决方案是 share()+filter()，比如

I'm new to reactive programming, and the only solution I come up with is share()+filter(), like

val shared = flux.share();
shared.filter(x -> x.tag=='a').subscribe(a -> consumeA(a));
shared.filter(x -> x.tag=='b').subscribe(b -> consumeB(b));
shared.filter(x -> x.tag=='c').subscribe(c -> consumeC(c));

这是最好的解决方案，还是有更好的范式来解决这个问题?

Is this the best solution, or is there any better paradigm for this problem?

推荐答案

如果组数比较少，那么可以使用 Flux.groupBy 引用在 项目反应器文档

If the number of groups is fairly low, then you can use Flux.groupBy referenced in the project reactor docs

例如:

Flux<String> flux = Flux.just("a1", "b1", "c1", "a2", "b2", "c2")
        .groupBy(s -> s.charAt(0))
        .concatMap(groupedFlux -> groupedFlux
                .startWith("Group " + groupedFlux.key()));

StepVerifier.create(flux)
        .expectNext("Group a", "a1", "a2")
        .expectNext("Group b", "b1", "b2")
        .expectNext("Group c", "c1", "c2")
        .verifyComplete();

您可以使用 groupedFlux.key() 来改变为每个组执行的操作.

You can use groupedFlux.key() to vary the operations performed for each group.

这篇关于在 Reactor 3 中将 Flux 拆分为多个 Flux 的最有效方法的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！