在 Reactor 2.0 中处理流过滤器的正确方法是什么？

Question

我正在使用 Reactor 2.0.0.M1，我正在尝试过滤 Stream。根据我的布尔运算的结果，我想继续使用一个或另一个流。 otherwise() 函数似乎可以做到这一点，但不清楚如何使用它。

我的信息流看起来像这样：

stream.filter(o -> o.isValid());

为了处理o.isValid() 为真的情况，我的理解是我可以直接调用.map() 继续顺流。

为了处理o.isValid() 为假的情况，我可以访问备用.otherwise() 流。

但似乎没有or() 或类似方法，因此似乎无法以完全流畅的方式配置两个流。

我能想到的最好的方法是这样的：

FilterAction<Object> filterAction = stream.filter(o -> o.isValid());

// Returns a 'true' Stream, which might additional operations
filterAction
    .map(o -> trueOperation1(o))
    .map(o -> trueOperation2(o));

// Returns a 'false' Stream, which might different additional operations
filterAction.otherwise()
    .map(o -> falseOperation1(o))
    .map(o -> falseOperation2(o));

这真的是最好的方法吗？

Answer 1

我通过使用 groupBy() 和 flatMap() 解决了这个问题。

这是一个例子：

// your initial stream
Broadcaster<Object> stream = Streams.<Object>broadcast(environment);

stream
    .groupBy(o -> o.isValid())
    .flatMap(groupedStream -> {
        if (groupedStream.key()) {
            return groupedStream.map(o -> trueOperation(o));
        } else {
            return groupedStream.map(o -> falseOperation(o));
    }
    .map(o -> additionalOperations();

这里发生的是 groupBy() 将您的 Stream 转换为 Stream<GroupedStream<O>>。换句话说，对象流的流。每个内部流包含一组对象，这些对象由 groupBy() 调用中的操作分桶。就我而言，我已将对象过滤到 true 和 false 存储桶中。

接下来，flatMap() 获取多个流，对其进行处理，然后将输出展平为单个 Stream<Object>。在 flatMap() 中，可以查看 Stream 的 key()，并根据 key() 对 Stream 进行额外的操作。

然后在 flatMap() 完成后，你又拥有了一个 Stream，并且可以进行任何你想要的后期处理。

Answer 2

看起来像你想要的

stream.filter(o -> {
  if (o.isValid()) {
    return trueOperation(o);
  } else {
    return falseOperation(o);
  }
});