android - 使用 RxJava 缓存网络调用一段时间

Question

我正在使用 Retorfit + RxJava2 创建一个网络，我想将响应缓存 30 秒。在 30 秒间隔后进行的任何调用都应该从服务器获得最新结果。我尝试使用 Replay 运算符执行此操作，但每次我调用订阅时它仍然会进行网络调用。我不是 RxJava 方面的专家，所以我对使用 Replay 进行缓存的理解可能是错误的。

public Observable<Name> getName() {
        return retrofitBuilder.getName()
                .subscribeOn(Schedulers.io())
                .replay(30, TimeUnit.SECONDS,Schedulers.io())
                .autoConnect();
    }

我这样调用上面的代码:

 service.getName()
        .subscribe(new Consumer<Name>()
            {
                @Override
                public void accept(Name name) throws Exception
                {
                    Log.d("getName", "Name: " + name.toString());
                }
            }
            , new Consumer<Throwable>()
            {
                @Override
                public void accept(Throwable throwable) throws Exception
                {
                    Log.d("getName", throwable.getMessage());
                }
            });

更新:如果我没有清楚地解释我的问题，我深表歉意。我想要的是对特定请求进行缓存，而不是将其缓存在 HttpClient 级别，后者将缓存策略应用于通过它发出的所有请求。最后我想在需要的时候为不同的请求定义不同的缓存过期时间。并非我的所有请求都需要缓存一小段时间。我想知道我是否能做到这一点。

在此感谢您的帮助。

Answer 1

您的方法有两个问题:

1. 正如@drhr 所提到的，每次调用 service.getName() 时，您都在创建一个新的 Observable 您正在创建一个新的 Observable< 实例，你应该保留相同的重播实例，并在每次调用 service.getName() 时给同一个实例之外的调用者。
2. 即使您将返回相同的实例，重播 30 秒，也会重播源 Observable 在过去 30 秒内发出的序列，这意味着在缓存过期之后时间，您将一无所获，因为您的请求发生在 30 多秒前。这并不意味着 Observable 会在这段时间后自动重启。

为了缓存特定时间段，您基本上需要在缓存时间段后使缓存的响应无效，并在此时间段后执行新请求，这意味着您应该控制您的订阅，并在那里进行。
你可以用类似的东西来实现它:

public class CachedRequest<T> {

    private final AtomicBoolean expired = new AtomicBoolean(true);
    private final Observable<T> source;
    private final long cacheExpirationInterval;
    private final TimeUnit cacheExpirationUnit;
    private Observable<T> current;

    public CachedRequest(Observable<T> o, long cacheExpirationInterval,
                         TimeUnit cacheExpirationUnit) {
        source = o;
        current = o;
        this.cacheExpirationInterval = cacheExpirationInterval;
        this.cacheExpirationUnit = cacheExpirationUnit;
    }

    private Observable<T> getCachedObservable() {
        return Observable.defer(() -> {
            if (expired.compareAndSet(true, false)) {
                current = source.cache();
                Observable.timer(cacheExpirationInterval, cacheExpirationUnit)                          
                        .subscribe(aLong -> expired.set(true));
            }
            return current;
        });
    }
}

使用 defer 你可以根据缓存过期状态返回正确的 Observable，所以在缓存过期期间发生的每个订阅都将被缓存 Observable(使用 cache( )) - 表示请求将只执行一次。缓存过期后，额外的订阅将触发新的请求，并设置一个新的计时器来重置缓存过期。