RxJava实现单次过滤分流数据记录：替代双filter方案探讨

我希望用响应式编程处理数据记录，需要先按规则过滤记录：

• 坏记录仅由RejectedRecordsSubscriber消费
• 好记录进入后续RecordProcessor等处理环节

处理流程架构如下：

| -> RejectedRecordsSubscriber
                                            |
RecordPublisher -> RecordMatcherProcessor -> RecordProcessor -> ...

由于过滤操作成本较高，我不想使用两个filter()算子（避免重复计算过滤逻辑），希望单次过滤后将记录分发至对应订阅者。请问在RxJava中如何实现该需求？groupBy是唯一方案吗？

注：我用Java Flow API编写的POC中已知订阅者类型，可直接分发至对应订阅者。

在RxJava里，groupBy不是唯一方案，有几种更贴合你需求的实现方式，能避免重复执行过滤逻辑：

1. 使用partition()算子（推荐） #

RxJava提供的partition()算子本质是对groupBy的封装，但语义更贴合“二分分发”场景——直接把数据流拆分成两个Flowable：一个是符合条件的好记录流，一个是不符合的坏记录流，且全程只执行一次过滤判断。

示例代码：

// 假设你的过滤判断逻辑是isGoodRecord(Record)
Pair<Flowable<Record>, Flowable<Record>> partitioned = recordFlowable.partition(this::isGoodRecord);

// 好记录流交给RecordProcessor处理
partitioned.first.subscribe(recordProcessor);
// 坏记录流交给RejectedRecordsSubscriber处理
partitioned.second.subscribe(rejectedRecordsSubscriber);

这种方式代码简洁直观，完全满足“单次过滤分发”的核心需求。

2. 自定义FlowableProcessor（灵活度更高） #

如果需要更定制化的分发逻辑（比如后续可能扩展更多记录类型），可以自己实现FlowableProcessor，在onNext()中直接判断并分发数据到指定订阅者，和你用Java Flow API写的POC思路完全对齐。

示例代码：

public class RecordMatcherProcessor extends FlowableProcessor<Record> {
    private final Subscriber<Record> goodRecordSubscriber;
    private final Subscriber<Record> badRecordSubscriber;

    public RecordMatcherProcessor(Subscriber<Record> good, Subscriber<Record> bad) {
        this.goodRecordSubscriber = good;
        this.badRecordSubscriber = bad;
    }

    @Override
    protected void subscribeActual(Subscriber<? super Record> s) {
        s.onSubscribe(new BooleanSubscription());
    }

    @Override
    public void onNext(Record record) {
        // 单次判断后直接分发
        if (isGoodRecord(record)) {
            goodRecordSubscriber.onNext(record);
        } else {
            badRecordSubscriber.onNext(record);
        }
    }

    @Override
    public void onError(Throwable t) {
        goodRecordSubscriber.onError(t);
        badRecordSubscriber.onError(t);
    }

    @Override
    public void onComplete() {
        goodRecordSubscriber.onComplete();
        badRecordSubscriber.onComplete();
    }

    @Override
    public boolean hasSubscribers() {
        return !(goodRecordSubscriber instanceof Subscription && ((Subscription) goodRecordSubscriber).isCancelled())
                || !(badRecordSubscriber instanceof Subscription && ((Subscription) badRecordSubscriber).isCancelled());
    }
}

这种方式完全由你控制分发逻辑，适合复杂业务场景。

3. groupBy方案（通用分组场景） #

如果后续需要拆分出更多类型的记录（不止好坏两类），groupBy是更通用的选择，它可以按自定义分组键拆分数据流：

示例代码：

recordFlowable.groupBy(record -> isGoodRecord(record) ? "GOOD" : "BAD")
        .subscribe(groupedFlowable -> {
            if ("GOOD".equals(groupedFlowable.getKey())) {
                groupedFlowable.subscribe(recordProcessor);
            } else {
                groupedFlowable.subscribe(rejectedRecordsSubscriber);
            }
        });

不过仅针对“好坏二分”的场景，partition()比groupBy更简洁、语义更清晰。

内容的提问来源于stack exchange，提问作者MiniScalope