java - 与 Flux 相交操作 - Project Reactor

Question

假设我有两个 var a = Flux.just("A", "B", "C")和 var b = Flux.just("B", "C", "D")
我希望能够将两个变量相交，结果应该相当于一个集合相交

类似 a.intersect(b)或 Flux.intersect(a, b)这将导致(通量)["B", "C"]
我找不到任何可以执行此操作的操作，有什么想法吗？

Answer 1

你可以像这样使用 join、filter、map 和 groupBy

    //Join fluxes in tuple
    a.join(b,s -> Flux.never(), s-> Flux.never(),Tuples::of)
            //Filter out matching
            .filter(t -> t.getT1().equals(t.getT2()))
            //Revert to single value
            .map(Tuple2::getT1)
            //Remove duplicates
            .groupBy(f -> f)
            .map(GroupedFlux::key)
            .subscribe(System.out::println);

结果是对每个订阅者进行单一订阅，并且还可以与欺骗者一起使用。

或者你可以编写自己的 intersect 方法

public <T> Flux<T> intersect(Flux<T> f1,Flux<T> f2){
    return f1.join(f2,f ->Flux.never(),f-> Flux.never(),Tuples::of)
            .filter(t -> t.getT1().equals(t.getT2()))
            .map(Tuple2::getT1)
            .groupBy(f -> f)
            .map(GroupedFlux::key);
}

//Use on it's own
intersect(a,b).subscribe(System.out::println)

//Or with existing flux
a.transform(f -> intersect(a,f)).subscribe(System.out::println)