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文檔中沒有這樣的保證。

如果事先不知道所有的事件類型，那麼這將不起作用。