使用 RxJava 管理相同的请求

Question

假设我有一个从单独线程上的给定链接获取图像的提取器。然后图像将被缓存在内存中。一旦图像已经被缓存，获取器将不会重新获取链接。 fetcher 被认为是一个 Observable。可能有很多订阅者向获取者索要图片。在第一个订阅者订阅 fetcher 后，fetcher 将拍摄网络。但是，如果有第二个订阅者来订阅，那么获取器不应该在它之前已经获取一个请求时再发送另一个请求。之后，如果获取完成，两个订阅者都将获得图像。现在，如果有第三个订阅者来了，提取器将立即发出图像。

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

Answer 1

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

答案是cache() 运算符：

“记住 Observable 发出的项目的序列，并将相同的序列发送给未来的订阅者”

来自：https://github.com/ReactiveX/RxJava/wiki/Observable-Utility-Operators

所以，下面的Observable 应该只获取一次图片，不管Subscribers 如何订阅它：

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

Answer 2

这可以通过 ConcurrentMap 和 AsyncSubject 来完成：

import java.awt.image.BufferedImage;
import java.io.*;
import java.net.URL;
import java.util.concurrent.*;

import javax.imageio.ImageIO;

import rx.*;
import rx.Scheduler.Worker;
import rx.schedulers.Schedulers;
import rx.subjects.AsyncSubject;


public class ObservableImageCache {
    final ConcurrentMap<String, AsyncSubject<BufferedImage>> image = 
        new ConcurrentHashMap<>();
    public Observable<BufferedImage> get(String url) {
        AsyncSubject<BufferedImage> result = image.get(url);
        if (result == null) {
            result = AsyncSubject.create();
            AsyncSubject<BufferedImage> existing = image.putIfAbsent(url, result);
            if (existing == null) {
                System.out.println("Debug: Downloading " + url);
                AsyncSubject<BufferedImage> a = result;
                Worker w = Schedulers.io().createWorker();
                w.schedule(() -> {
                    try {
                        Thread.sleep(500); // for demo
                        URL u = new URL(url);

                        try (InputStream openStream = u.openStream()) {
                            a.onNext(ImageIO.read(openStream));
                        }
                        a.onCompleted();
                    } catch (IOException | InterruptedException ex) {
                        a.onError(ex);
                    } finally {
                        w.unsubscribe();
                    }
                });
            } else {
                result = existing;
            }
        }
        return result;
    }
    public static void main(String[] args) throws Exception {
        ObservableImageCache cache = new ObservableImageCache();
        CountDownLatch cdl = new CountDownLatch(4);

        Observable<BufferedImage> img1 = cache.get("https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/create.png");
        System.out.println("Subscribing for IMG1");
        img1.subscribe(e -> System.out.println("IMG1: " + e.getWidth() + "x" + e.getHeight()), Throwable::printStackTrace, cdl::countDown);
        Thread.sleep(500);
        Observable<BufferedImage> img2 = cache.get("https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/create.png");
        System.out.println("Subscribing for IMG2");
        img2.subscribe(e -> System.out.println("IMG2: " + e.getWidth() + "x" + e.getHeight()), Throwable::printStackTrace, cdl::countDown);

        Observable<BufferedImage> img3 = cache.get("https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/amb.png");
        Observable<BufferedImage> img4 = cache.get("https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/amb.png");

        Thread.sleep(500);

        System.out.println("Subscribing for IMG3");
        img3.subscribe(e -> System.out.println("IMG3: " + e.getWidth() + "x" + e.getHeight()), Throwable::printStackTrace, cdl::countDown);
        Thread.sleep(1000);
        System.out.println("-> Should be immediate: ");
        System.out.println("Subscribing for IMG4");
        img4.subscribe(e -> System.out.println("IMG4: " + e.getWidth() + "x" + e.getHeight()), Throwable::printStackTrace, cdl::countDown);

        cdl.await();
    }
}

我正在使用 ConcurrentMap 的 putIfAbsent 来确保新 url 只触发一次下载；其他所有人都将收到相同的 AsyncSubject，他们可以“等待”并在可用后立即获取数据。通常，您希望通过使用自定义调度程序来限制并发下载的数量。

Answer 3

看看ConnectableObservable 和.replay() 方法。

我目前正在使用这是我的片段来处理方向变化：

Fragment 的 onCreate：

ConnectableObservable<MyThing> connectableObservable = 
    retrofitService.fetchMyThing()
        .map(...)
        .replay();

connectableObservable.connect(); // this starts the actual network call

Fragment 的 onCreateView：

Subscription subscription = connectableObservable
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(mything -> dosomething());

发生的情况是我只发出 1 个网络请求，并且任何订阅者都会（最终/立即）获得该响应。