使用RxJava管理相同的请求

Question

我如何用RxJava方法实现上面的场景？我希望利用某种现有的运算符，以声明性更强的方式组合它们，最重要的是，避免同步、锁和原子的开销。

Answer 1

问题中的线索是：“假设我有一个提取器，它从一个独立线程上的给定链接中提取一个图像。然后图像将被缓存在内存中。”

答案是cache()运算符：

记住由可观察者发出的项目序列，并向未来的订户发出相同的序列

因此，无论订阅者如何订阅映像，以下可观察的应该只获取映像一次：

Observable<Bitmap> cachedBitmap = fetchBitmapFrom(url).cache();

编辑：

我认为下面的示例证明了上游Observable只订阅了一次，即使在Observable发出任何内容之前有多个订阅。对于网络请求也应该如此。

package com.example;

import rx.Observable;
import rx.Subscriber;
import rx.schedulers.Schedulers;

public class SimpleCacheTest {

    public static void main(String[] args) {
        final Observable<Integer> cachedSomething = getSomething().cache();

        System.out.println("before first subscription");
        cachedSomething.subscribe(new SimpleLoggingSubscriber<Integer>("1"));

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("before second subscription");
        cachedSomething.subscribe(new SimpleLoggingSubscriber<Integer>("2"));

        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("quit");
    }

    private static class SimpleLoggingSubscriber<T> extends Subscriber<T> {

        private final String tag;

        public SimpleLoggingSubscriber(final String tag) {
            this.tag = tag;
        }
        @Override
        public void onCompleted() {
            System.out.println("onCompleted (" + tag + ")");
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("onError (" + tag + ")");
        }

        @Override
        public void onNext(T t) {
            System.out.println("onNext (" + tag + "): " + t);
        }
    }

    private static Observable<Integer> getSomething() {
        return Observable.create(new Observable.OnSubscribe<Integer>(){

            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                System.out.println("going to sleep now...");
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                subscriber.onNext(1);
                subscriber.onCompleted();
            }
        }).subscribeOn(Schedulers.io());
    }
}

before first subscription
going to sleep now...
before second subscription
onNext (1): 1
onNext (2): 1
onCompleted (1)
onCompleted (2)
quit