如何通过将两个 Flux 的值配对到一个元组中来组合发布者？

Question

假设我有两个 Flux 如下：

    Flux<Integer> f1 = Flux.just(10,20,30,40);
    Flux<Integer> f2 = Flux.just(100,200,300,400);

现在我想要的是将这些通量组合成单个通量或两个通量的元组，这将在单个通量中包含两个通量的元素。

我使用 zipwith 方法尝试了以下操作：

  Flux<Integer, Integer> zipped = f1.zipWith(f2,
                                            (one, two) ->  one + "," +two)
                                    .subscribe();

但这给出了编译时错误：

 Incorrect number of arguments for type Flux<T>; it cannot be parameterized with arguments <Integer, Integer>

我怎样才能做到这一点？请提出建议。

Answer 1

Flux 只有一个类型参数，所以 Flux<Integer,Integer> 是不可能的，我不确定你想用 one + "," + two 实现什么，但是这个表达式的类型是 String .

所以，本质上，您是将两个整数映射到一个字符串，因此 zipped 的类型应该是 Flux<String>。

或者，您可以映射到您自己创建的特殊元组 class（或者可能来自您正在使用的库）：

public class Pair<A,B> {
    private final A first;

    private final B second;

    public Pair(A first, B second) {
        this.first = first;
        this.second = second;
    }

    public A getFirst() {
        return first;
    }

    public B getSecond() {
        return second;
    }
}

然后您可以将其映射如下：

Flux<Integer> f1 = Flux.just(10,20,30,40);
Flux<Integer> f2 = Flux.just(100,200,300,400);

Flux<Pair<Integer,Integer>> result = f1.zipWith(f2,
    (one, two) ->  new Pair<>(one, two));

或更短：

Flux<Pair<Integer,Integer>> result = f1.zipWith(f2, Pair::new);

如何通过将两个 Flux 的值配对到一个元组中来组合发布者？

How to combine publishers by pairing values from two Flux into a Tuple?

java

publisher

publish-subscribe

project-reactor

reactive-streams