RxJava: почему одни и те же преобразования пересчитываются для каждой ветки наблюдаемых?

Observable лениво выполняется каждый раз, когда на него подписываются (либо явно, либо неявно через композицию).

Этот код показывает, как источник излучает для a, b и c:

    Observable<Integer> d = Observable.range(1, 10)
            .doOnNext(i -> System.out.println("Emitted from source: " + i));
    Observable<Integer> c = d.map(t -> t + 1);
    Observable<Integer> a = c.map(t -> t + 2);
    Observable<Integer> b = c.map(t -> t + 3);

    a.forEach(i -> System.out.println("a: " + i));
    b.forEach(i -> System.out.println("b: " + i));
    c.forEach(i -> System.out.println("c: " + i));

Если вы в порядке с буферизацией (кешированием) результата, то это так же просто, как использовать оператор .cache() для достижения этого.

    Observable<Integer> d = Observable.range(1, 10)
            .doOnNext(i -> System.out.println("Emitted from source: " + i))
            .cache();

    Observable<Integer> c = d.map(t -> t + 1);
    Observable<Integer> a = c.map(t -> t + 2);
    Observable<Integer> b = c.map(t -> t + 3);

    a.forEach(i -> System.out.println("a: " + i));
    b.forEach(i -> System.out.println("b: " + i));
    c.forEach(i -> System.out.println("c: " + i));

Добавление .cache() к источнику делает его испускаемым только один раз и может быть подписано много раз.

Для больших или бесконечных источников данных кэширование не подходит, поэтому многоадресная рассылка является решением, гарантирующим, что источник передает только один раз.

Операторы publish() и share() — хорошее место для начала, но для простоты и, поскольку это синхронный пример, я покажу перегрузку publish(function), которую часто проще всего использовать.

    Observable<Integer> d = Observable.range(1, 10)
            .doOnNext(i -> System.out.println("Emitted from source: " + i))
            .publish(oi -> {
                Observable<Integer> c = oi.map(t -> t + 1);
                Observable<Integer> a = c.map(t -> t + 2);
                Observable<Integer> b = c.map(t -> t + 3);

                return Observable.merge(a, b, c);
            });

    d.forEach(System.out::println);

Если a, b, c нужны по отдельности, мы можем подключить все и «подключить» источник, когда он будет готов:

private static void publishWithConnect() {
    ConnectableObservable<Integer> d = Observable.range(1, 10)
            .doOnNext(i -> System.out.println("Emitted from source: " + i))
            .publish();

    Observable<Integer> c = d.map(t -> t + 1);
    Observable<Integer> a = c.map(t -> t + 2);
    Observable<Integer> b = c.map(t -> t + 3);

    a.forEach(i -> System.out.println("a: " + i));
    b.forEach(i -> System.out.println("b: " + i));
    c.forEach(i -> System.out.println("c: " + i));

    // now that we've wired up everything we can connect the source
    d.connect();
}

Или, если источник асинхронный, мы можем использовать refCounting:

    Observable<Integer> d = Observable.range(1, 10)
            .doOnNext(i -> System.out.println("Emitted from source: " + i))
            .subscribeOn(Schedulers.computation())
            .share();

Однако refCount (share — это перегрузка для его предоставления) допускает условия гонки, поэтому не гарантирует, что все подписчики получат первые значения. Обычно это требуется только для «горячих» потоков, на которые приходят и уходят подписчики. Для «холодного» источника, который мы хотим обеспечить всем, предпочтительным подходом являются предыдущие решения с cache() или publish()/publish(function).

Вы можете узнать больше здесь: https://github.com/ReactiveX/RxJava/wiki/Connectable-Observable-Operators