RXJava技术问题：如何让Observable II等待Observable I完成

Great question! This is a common scenario when dealing with dependent streams in Rx, and the key here is to share the initial cold observable in a way that it only runs once, while letting subsequent subscribers (like your pagination stream) wait for it to complete if it's still running, or proceed immediately if it's already done.

Step 1: Share and Cache the Initial Observable #

First, convert your cold Observable I into a shared, single-execution observable that caches its completion state (and data if you need to use it later). This ensures it runs exactly once, no matter how many times other streams subscribe to it.

RxJava Example: #

// Convert Observable I into a shared, cached observable
Observable<InitialData> sharedInitialData = observableI.cache();

RxJS Example: #

// Share and cache the initial observable, ensuring it runs once
const sharedInitialData$ = observableI$.pipe(
  shareReplay({ bufferSize: 1, refCount: false })
);

Why this works:

• cache() (RxJava) subscribes to the cold observable once, caches all emissions and the completion event, and replays them to every new subscriber.
• shareReplay with refCount: false (RxJS) keeps the subscription alive even when there are no active subscribers, guaranteeing the initial observable only executes once.

Step 2: Make Pagination Stream Wait for Initial Completion #

For your hot pagination stream (Observable II), use flatMap (or concatMap if you need to preserve pagination order) to first wait for the shared initial observable to complete, then execute your pagination logic.

RxJava Example: #

// Observable II is your hot pagination trigger stream (e.g., emits page numbers/requests)
observableII.flatMap(pageRequest -> 
    // Wait for initial data to complete (ignore the data if you don't need it)
    sharedInitialData.ignoreElements()
                     // Once initial load is done, fetch the requested page
                     .andThen(loadPage(pageRequest))
)
.subscribe(pageData -> {
    // Handle your paginated data here
});

RxJS Example: #

// Observable II is your hot pagination trigger stream
observableII$.pipe(
  switchMap(pageRequest => 
    // Wait for initial data to complete, then load the page
    concat(
      sharedInitialData$.pipe(ignoreElements()),
      loadPage$(pageRequest)
    )
  )
).subscribe(pageData => {
  // Process paginated results here
});

How This Covers Your Scenarios #

1. Scenario 1: Observable I is still running when Observable II starts
   The ignoreElements() (RxJava) or concat + ignoreElements() (RxJS) will pause the pagination logic until sharedInitialData finishes. Your pagination requests won't execute until the initial data is fully loaded.

2. Scenario 2: Observable I has already completed when Observable II starts
   The shared observable will immediately emit its cached completion event, so the wait step finishes instantly. Your pagination logic runs right away without any unnecessary delays.

3. Guaranteed Single Execution of Observable I
   The cache()/shareReplay({ refCount: false }) operators ensure Observable I is only subscribed to once—even if Observable II triggers multiple times or has multiple subscribers.

Alternative: Using AsyncSubject (RxJava) #

If you prefer more explicit control, you can use an AsyncSubject, which only emits the last value of the source observable when it completes:

AsyncSubject<InitialData> initialDataSubject = AsyncSubject.create();
// Subscribe the subject to Observable I (runs once)
observableI.subscribe(initialDataSubject);

// Use the subject in your pagination stream
observableII.flatMap(pageRequest -> 
    initialDataSubject.ignoreElements()
                     .andThen(loadPage(pageRequest))
)
.subscribe(...);

This achieves the same result as cache(), but cache() is more concise for this specific use case.

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