RxJava2：如何为多个Rx链复用指定操作符序列？

Great question! Reusing common operator sequences in RxJava is a fantastic way to keep your code DRY (Don’t Repeat Yourself) and easier to maintain. Here’s a clean, scalable approach to package that sequence for reuse across multiple Rx chains:

Step 1: Create a Reusable ObservableTransformer #

We’ll wrap your target operators in a generic ObservableTransformer—RxJava’s built-in tool for packaging reusable operator logic. This transformer will take an Observable<List<T>>, process it through your desired steps, and return an Observable<List<T>> (adjust the generic type T to match your actual model class):

import io.reactivex.Observable;
import io.reactivex.ObservableTransformer;
import java.util.Comparator;
import java.util.List;

public class RxUtils {
    // Generic transformer to flatten, sort, and convert back to a List
    public static <T> ObservableTransformer<List<T>, List<T>> flattenSortAndToList(Comparator<T> comparator) {
        return upstream -> upstream
            .flatMapIterable(items -> items) // Flatten the List into individual items
            .sorted(comparator) // Sort items using the provided comparator
            .toList() // Collect sorted items back into a List
            .toObservable(); // Convert Single<List<T>> back to Observable<List<T>>
    }
}

Step 2: Apply the Transformer in Your Rx Chain #

Replace the duplicated operators with the compose() method to inject your reusable sequence. This keeps your original chain concise and consistent:

compositeDisposable.add(manager.getObservable()
    .map(objects -> modelMapper.map(objects))
    .map(modelObjects -> { 
        cache.save(modelObjects); 
        return modelObjects; 
    })
    .compose(RxUtils.flattenSortAndToList(objectComparator)) // Reuse the operator sequence here
    .map(modelObjects -> viewModelMapper.map(modelObjects))
    .subscribe(this::onObjectsLoaded));

Key Benefits of This Approach #

• Efficiency: Using compose() applies the transformer directly to the upstream Observable, avoiding the overhead of nested Observables (unlike flatMap()).
• Flexibility: The generic <T> type lets you reuse this transformer with any model class, as long as you provide a matching Comparator<T>.
• Maintainability: If you ever need to adjust the sequence (e.g., add a filter, tweak sorting logic), you only update it once in the transformer instead of every Rx chain.

Alternative: For Single Chains #

If your upstream source returns a Single instead of an Observable, you can create a SingleTransformer variant:

import io.reactivex.Single;
import io.reactivex.SingleTransformer;
import java.util.Comparator;
import java.util.List;

public class RxUtils {
    public static <T> SingleTransformer<List<T>, List<T>> flattenSortAndToListSingle(Comparator<T> comparator) {
        return upstream -> upstream
            .flatMapObservable(Observable::fromIterable)
            .sorted(comparator)
            .toList();
    }
}

Use it with a Single chain like this:

compositeDisposable.add(manager.getSingle() // Assuming manager returns Single<List<...>>
    .map(objects -> modelMapper.map(objects))
    .map(modelObjects -> { 
        cache.save(modelObjects); 
        return modelObjects; 
    })
    .compose(RxUtils.flattenSortAndToListSingle(objectComparator))
    .map(modelObjects -> viewModelMapper.map(modelObjects))
    .subscribe(this::onObjectsLoaded));

内容的提问来源于stack exchange，提问作者Marius Andrei Rosu