RxJava -2 Observables可以随时接受更多Observables吗?

问题描述

我目前正在使用rx-java 2，并且有一个用例，其中单个Camel Route用户需要使用多个Observable.以此解决方案作为参考，我有一个部分可行的解决方案. RxJava-合并的Observable可以随时接受更多Observables吗?

I'm currently using rx-java 2 and have a use case where multiple Observables need to be consumed by single Camel Route subscriber. Using this solution as a reference, I have a partly working solution. RxJava - Merged Observable that accepts more Observables at any time?

我计划使用PublishProcessor<T>，它将订阅到一个骆驼反应流订阅者，然后维护一个ConcurrentHashSet<Flowable<T>>，在这里我可以动态添加新的Observable.
我目前无法使用PublishProcessor添加/管理Flowable<T>实例吗?我真的是rx java的新手，所以对您的帮助不胜感激！这是我到目前为止所拥有的:

I'm planning to use a PublishProcessor<T> that will be subscribed to one camel reactive stream subscriber and then maintain a ConcurrentHashSet<Flowable<T>> where I can dynamically add new Observable.
I'm currently stuck on how can I add/manage Flowable<T> instances with PublishProcessor? I'm really new to rx java, so any help is appreciated! This is what I have so far :

PublishProcessor<T> publishProcessor = PublishProcessor.create();
CamelReactiveStreamsService camelReactiveStreamsService = 
CamelReactiveStreams.get(camelContext);
Subscriber<T> subscriber = 
     camelReactiveStreamsService.streamSubscriber("t-class",T.class);
}
Set<Flowable<T>> flowableSet = Collections.newSetFromMap(new ConcurrentHashMap<Flowable<T>, Boolean>());

public void add(Flowable<T> flowableOrder){
    flowableSet.add(flowableOrder);
}

public void subscribe(){
    publishProcessor.flatMap(x -> flowableSet.forEach(// TODO)
    }) .subscribe(subscriber);
}

推荐答案

您可以拥有一个Processor并订阅多个可观察的流.您需要在添加和删除可观察对象时通过添加和删除订阅来管理订阅.

You can have a single Processor and subscribe to more than one observable stream. You would need to manage the subscriptions by adding and removing them as you add and remove observables.

PublishProcessor<T> publishProcessor = PublishProcessor.create();

Map<Flowable<T>, Disposable> subscriptions = new ConcurrentHashMap<>();

void addObservable( Flowable<T> flowable ) {
  subscriptions.computeIfAbsent( flowable, fkey -> 
    flowable.subscribe( publishProcessor ) );
}
void removeObservable( Flowable<T> flowable ) {
  Disposable d = subscriptions.remove( flowable );
  if ( d != null ) {
    d.dispose();
  }
}
void close() {
  for ( Disposable d: subscriptions.values() ) {
    d.dispose();
  }
}

使用flowable作为地图的键，并添加或删除订阅.

Use the flowable as the key to the map, and add or remove subscriptions.

这篇关于RxJava -2 Observables可以随时接受更多Observables吗?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！