RxJava2链式调用不同类型Observable的实现及问题咨询

Hey there! Let's walk through a clean, maintainable way to implement this RxJava2 chain that meets all your requirements. I'll break it down step by step so you can adapt it to your actual code.

First, Define Your Core Observables #

Start by creating three Observable instances that handle each specific task. Make sure to offload all heavy work (data processing, network calls) to background threads with subscribeOn(Schedulers.io()):

// 1. Handles data processing (emits one processed data object then completes)
Observable<ProcessedData> processData() {
    return Observable.create(emitter -> {
        // Replace with your actual data processing logic
        ProcessedData processedData = new ProcessedData();
        emitter.onNext(processedData);
        emitter.onComplete();
    }).subscribeOn(Schedulers.io());
}

// 2. Uploads to server1, emits progress percentages (0-100) then completes
Observable<Integer> uploadToServer1(ProcessedData data) {
    return Observable.create(emitter -> {
        // Replace with real upload logic that tracks progress
        int[] progressSteps = {0, 25, 50, 75, 100};
        for (int progress : progressSteps) {
            if (!emitter.isDisposed()) {
                emitter.onNext(progress);
                Thread.sleep(500); // Simulate network delay
            }
        }
        emitter.onComplete();
    }).subscribeOn(Schedulers.io());
}

// 3. Notifies server2 that upload is done (emits a completion signal)
Observable<Void> notifyServer2() {
    return Observable.create(emitter -> {
        // Replace with your actual server2 notification call
        Thread.sleep(300); // Simulate network call
        emitter.onNext(null);
        emitter.onComplete();
    }).subscribeOn(Schedulers.io());
}

Chain Them Together & Handle UI Updates #

The key here is to chain these Observables while preserving progress events, then switch to the main thread for UI updates. To avoid messy type checks, use an abstract event class (or sealed class in Kotlin) for type safety:

Step 1: Create an Event Class (Type Safety)

In Java, use an abstract base class with concrete subclasses to categorize events:

abstract class UploadEvent {}

class ProgressEvent extends UploadEvent {
    public final int progress;
    public ProgressEvent(int progress) { this.progress = progress; }
}

class NotifyCompletedEvent extends UploadEvent {}

Step 2: Build the Rx Chain

Now, chain the Observables, map events to our event class, and handle UI updates:

// In your Activity
private CompositeDisposable disposables = new CompositeDisposable();

private void startUploadFlow() {
    Disposable uploadFlow = processData()
            .flatMap(processedData -> 
                // First, emit progress events from the upload
                uploadToServer1(processedData)
                        .map(ProgressEvent::new)
                        // After upload completes, trigger server2 notification and emit completion event
                        .concatWith(notifyServer2().map(ignored -> new NotifyCompletedEvent()))
            )
            // Switch to main thread for all UI operations
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(
                    event -> {
                        // Handle each event type
                        if (event instanceof ProgressEvent) {
                            int progress = ((ProgressEvent) event).progress;
                            // Update your progress bar or text view
                            updateProgressUI(progress);
                        } else if (event instanceof NotifyCompletedEvent) {
                            // Show success state (e.g., toast, success screen)
                            showSuccessState();
                        }
                    },
                    error -> {
                        // Handle ANY error from any Observable in the chain
                        showErrorUI(error.getMessage());
                    }
            );

    // Add to composite disposable to clean up later
    disposables.add(uploadFlow);
}

// Don't forget to clean up subscriptions to avoid memory leaks
@Override
protected void onDestroy() {
    super.onDestroy();
    disposables.dispose();
}

Key Notes to Remember #

• Thread Safety: All background work uses subscribeOn(Schedulers.io()), and UI updates are routed to the main thread via observeOn(AndroidSchedulers.mainThread()).
• Error Handling: Any exception thrown by processData(), uploadToServer1(), or notifyServer2() will trigger the onError callback—you only need one error handler for the entire chain.
• Memory Leaks: Use CompositeDisposable to dispose of subscriptions when the Activity is destroyed, preventing memory leaks.
• Flexibility: If you prefer a more concise (slightly less type-safe) approach, you can skip the event class and check emitted object types directly:

  .subscribe(
          result -> {
              if (result instanceof Integer) {
                  updateProgressUI((Integer) result);
              } else {
                  showSuccessState();
              }
          },
          error -> showErrorUI(error.getMessage())
  );
  

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