ReplaySubject不同的元素

Question

我想用RxJava实现一个EventBus，我需要粘性事件。我知道我可以使用BehaviorSubject，但它只缓存最后发出的项目，而我想缓存所有不同类型的事件（类名）。还有另一种选择 - ReplaySubject，但它有一个开销 - 它拥有所有发射的元素。 有没有一种方法来创建某种ReplaySubject，它只能通过类型元素保持唯一性？

Answer 1

我不认为有任何清洁的解决方案。但是，您可能可以完成这项工作。

1. 为每个事件类型创建一个BehaviorSubject<>。使用ConcurrentMap将每个传入事件分派到正确的Subject。
2. 按事件到达顺序维护这些主题的列表。
3. 当收到新的订阅时，将创建一个可观察的事件，即所有主题的合并，第一个主题列表已经收到事件，第二个主题列表中的事件没有任何顺序。

这是一些代码，可能会澄清上述情况。未经测试。

// The subscription operation will perform a merge of the two lists 
Map<EventType, BehaviorSubject<Event>> map = new ConcurrentHashMap<>(); 
List<BehaviorSubject<Event>> listOfUnseenEvents = new ArrayList<>(); 
List<BehaviorSubject<Event>> listOfSeenEvents = new ArrayList<>(); 
// ... 
listOfUnseenEvents = map.values().asList(); 

public Observable<Event> busSub() { 
    List<BehaviorSubject<Event>> allEvents = new ArrayList<>(); 
    synchronized (map) { 
    allEvents.addAll(listOfSeenEvents); 
    allEvents.addAll(listOfUnseenEvents); 
    } 
    return Observable.merge(allEvents); 
} 

// receive an event and dispatch it 
eventSource 
    .subscribe(event -> processEvent(event)); 

public void processEvent(Event event) { 
    BehaviorSubject<Event> eSubject = map.get(event.getEventType()); 
    synchronized (map) { 
     if (containsEventType(listOfSeenEvents, event.getEventType())) { 
     removeEventType(listOfSeenEvents, event.getEventType()); 
     } else { 
     removeEventType(listOfUnseenEvents, event.getEventType()); 
     } 
     listOfSeenEvents.add(eSubject); 
    } 
    eSubject.onNext(event); 
} 

请注意，此代码利用了一个事实，即merge()将订阅每个给定的观测，以便优势。在RxJava文档中没有这样的保证。

如果事先不知道所有的事件类型，那么这将不起作用。