La manière la plus efficace de diviser un flux en plusieurs flux dans le réacteur 3.

Question

Dans le réacteur 3, quel est le moyen le plus efficace de diviser un flux hétérogène en plusieurs flux par correspondance de motifs ? (Et les opérations ultérieures sur chaque flux peuvent être très différentes)

Par exemple,

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

Je suis novice en matière de programmation réactive, et la seule solution que j'ai trouvée est la suivante share() + filter() comme

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

Est-ce la meilleure solution, ou existe-t-il un meilleur paradigme pour ce problème ?

Answer 1

Si le nombre de groupes est assez faible, alors vous pouvez utiliser Flux.groupBy référencé dans le docs sur le réacteur du projet

Par exemple :

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

Vous pouvez utiliser groupedFlux.key() pour varier les opérations effectuées pour chaque groupe.