RxJava 2：如何实现可随时添加新Observable的合并Observable？

嘿，这个场景我之前在做Camel和RxJava集成的时候刚好碰过，你的思路完全找对了方向——用PublishProcessor作为统一出口对接Camel的响应式订阅者，再靠ConcurrentHashSet<Flowable<T>>来动态管理新增的Observable，下面我给你细化这个实现，补全关键的细节和踩坑点：

核心实现思路拆解 #

1. 初始化核心管理组件 #

首先我们需要一个单例的桥接类，用来维护PublishProcessor和动态Flowable集合，保证线程安全：

import io.reactivex.processors.PublishProcessor;
import io.reactivex.Flowable;
import java.util.concurrent.ConcurrentHashSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

// 单例桥接类，负责RxJava和Camel的事件转发
public class RxEventCamelBridge<T> {
    private static final Logger log = LoggerFactory.getLogger(RxEventCamelBridge.class);
    
    // 对接Camel的统一输出Processor，所有事件都会转发到这里
    private final PublishProcessor<T> eventProcessor = PublishProcessor.create();
    // 维护当前活跃的Flowable集合，用ConcurrentHashSet保证多线程下的安全操作
    private final ConcurrentHashSet<Flowable<T>> activeFlowables = new ConcurrentHashSet<>();

    // 暴露给Camel的订阅入口，用hide()避免外部直接操作Processor的生命周期方法
    public Flowable<T> getCamelSubscribableFlowable() {
        return eventProcessor.hide();
    }
}

2. 动态添加Observable/Flowable #

接下来实现动态添加数据源的方法，同时处理单个Flowable的生命周期（完成后自动从集合移除）：

// 动态添加Flowable数据源
public void addDataSource(Flowable<T> newFlowable) {
    // 避免重复添加同一个Flowable
    if (activeFlowables.add(newFlowable)) {
        // 订阅该Flowable，将事件转发到统一Processor
        newFlowable.subscribe(
            eventProcessor::onNext,
            error -> {
                // 这里可以根据业务选择是否将错误转发到Processor
                // 如果转发，Processor会终止所有订阅，所以建议单独处理单个数据源的错误
                log.error("Error from data source: {}", error.getMessage(), error);
                activeFlowables.remove(newFlowable);
            },
            () -> {
                // 数据源完成后，自动从集合中移除
                log.info("Data source completed, removing from active set");
                activeFlowables.remove(newFlowable);
            }
        );
    }
}

// 支持Observable转Flowable（适配RxJava2的背压机制，Camel更友好）
public void addDataSource(Observable<T> newObservable) {
    // 根据业务场景选择合适的背压策略：BUFFER/DROP/LATEST
    addDataSource(newObservable.toFlowable(BackpressureStrategy.BUFFER));
}

3. 对接Apache Camel Route #

最后把桥接类的Flowable接入Camel Route，用RxJava2组件消费事件：

import org.apache.camel.builder.RouteBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class RxEventConsumerRoute extends RouteBuilder {

    private final RxEventCamelBridge<YourBusinessType> eventBridge;

    @Autowired
    public RxEventConsumerRoute(RxEventCamelBridge<YourBusinessType> eventBridge) {
        this.eventBridge = eventBridge;
    }

    @Override
    public void configure() throws Exception {
        // 从RxJava2 Flowable订阅事件，转发到后续业务逻辑
        from("rxjava2:flowable?flowable=#camelEventFlowable")
            .routeId("RxEventToCamelRoute")
            .log("Received Rx event: ${body}")
            // 这里替换成你的业务处理逻辑，比如转发到数据库、消息队列等
            .to("direct:businessProcessingPipeline");
    }
}

// 在Spring容器中注册必要的Bean
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class CamelRxConfig {

    @Bean
    public RxEventCamelBridge<YourBusinessType> rxEventCamelBridge() {
        return new RxEventCamelBridge<>();
    }

    @Bean(name = "camelEventFlowable")
    public Flowable<YourBusinessType> camelEventFlowable(RxEventCamelBridge<YourBusinessType> bridge) {
        return bridge.getCamelSubscribableFlowable();
    }
}

关键注意事项 #

• 线程安全：必须使用ConcurrentHashSet（或ConcurrentHashMap如果需要保存Disposable），因为添加/移除数据源的操作大概率在多线程环境下进行（比如业务线程动态新增数据源）。
• 背压处理：Camel的RxJava2组件默认支持背压，但在将Observable转为Flowable时，一定要根据业务场景选择合适的BackpressureStrategy：
  • BUFFER：缓存所有事件（适合不能丢数据的场景，但要注意内存溢出）
  • DROP：丢弃最新的事件（适合允许丢数据的高并发场景）
  • LATEST：只保留最新的事件（适合只关心最新状态的场景）
• 错误隔离：单个数据源的错误如果直接转发到PublishProcessor，会导致整个Processor终止（RxJava的规则），所以建议在订阅时单独处理错误，避免影响其他数据源。
• 资源清理：除了数据源自动完成后移除，建议提供主动移除的方法（如果需要手动停止某个数据源），此时可以把集合改成ConcurrentHashMap<Flowable<T>, Disposable>，保存每个订阅的Disposable，方便主动取消订阅：

// 优化后的主动移除方法
private final ConcurrentHashMap<Flowable<T>, Disposable> disposableMap = new ConcurrentHashMap<>();

public void removeDataSource(Flowable<T> flowable) {
    Disposable disposable = disposableMap.remove(flowable);
    if (disposable != null && !disposable.isDisposed()) {
        disposable.dispose();
        log.info("Manually stopped data source");
    }
}

这样整个方案就能完美实现动态添加多个Observable/Flowable，统一由单个Camel Route订阅消费的需求了，你可以根据自己的业务类型替换YourBusinessType，调整错误处理和背压策略。

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