RxJava 一个发布者的多个消费者

Question

我正在编写某种带有缓存的中间件 HTTP 代理。工作流程是：

1. 客户端向此代理请求资源
2. 如果缓存中存在资源，则代理返回它
3. 如果未找到资源，则代理获取远程资源并返回给用户。代理在数据加载时将此资源保存到缓存中。

我的接口有用于远程资源的Publisher<ByteBuffer> 流，接受Publisher<ByteBuffer> 保存的缓存，以及接受Publisher<ByteBuffer> 作为响应的客户端连接：

// remote resource
interface Resource {
  Publisher<ByteBuffer> fetch();
}

// cache
interface Cache {
  Completable save(Publisher<ByteBuffer> data);
}

// clien response connection
interface Connection {
  Completable send(Publisher<ByteBuffer> data);
}

我的问题是我需要在向客户端发送响应时延迟保存这个字节缓冲区流以缓存，因此客户端应该负责从远程资源请求ByteByffer块， 不缓存。

我尝试使用Publisher::cache 方法，但这对我来说不是一个好的选择，因为它将所有接收到的数据保存在内存中，这是不可接受的，因为缓存的数据可能只有几 GB 大小。

作为一种解决方法，我创建了Subject，并由来自Resource 的下一个项目填充：

private final Cache cache;
private final Connection out;

Completable proxy(Resource res) {
  Subject<ByteBuffer> mirror = PublishSUbject.create();
  return Completable.mergeArray(
    out.send(res.fetch().doOnNext(mirror::onNext),
    cache.save(mirror.toFlowable(BackpressureStrategy.BUFFER))
  );
}

是否可以重复使用相同的 Publisher 而无需在内存中缓存项目，并且只有一个订阅者负责从发布者请求项目？

Answer 1

我可能遗漏了一些东西（添加了关于我的 Publisher 界面版本不同的评论）。

但是..这就是我在概念上会如何做这样的事情。

我将简化处理Integers的接口：

// remote resource
interface Resource {
  ConnectableObservable<Integer> fetch();
}

// cache
interface Cache {
  Completable save(Integer data);
}

// client response connection
interface Connection {
  Completable send(Integer data);
}

我会使用Observable::publish 创建一个ConnectableObservable 并建立两个订阅：

@Test
public void testProxy()
{
    // Override schedulers:
    TestScheduler s = new TestScheduler();
    
    RxJavaPlugins.setIoSchedulerHandler(
            scheduler -> s );
    RxJavaPlugins.setComputationSchedulerHandler(
            scheduler -> s );
    
    // Mock interfaces:
    Resource resource = () -> Observable.range( 1, 100 )
            .publish();
    
    Cache cache = data -> Completable.fromObservable( Observable.just( data )
                .delay( 100, TimeUnit.MILLISECONDS )
                .doOnNext( __ -> System.out.println( String.format( "Caching %d", data ))));
    
    Connection connection = data -> Completable.fromObservable( Observable.just( data )
                .delay( 500, TimeUnit.MILLISECONDS )
                .doOnNext( __ -> System.out.println( String.format( "Sending %d", data ))));
    
    // Subscribe to resource:
    ConnectableObservable<Integer> observable = resource.fetch();
    
    observable
        .observeOn( Schedulers.io() )
        .concatMapCompletable( data -> connection.send( data ))
        .subscribe();
    
    observable
        .observeOn( Schedulers.computation() )
        .concatMapCompletable( data -> cache.save( data ))
        .subscribe();
    
    observable.connect();
    
    // Simulate passage of time:
    s.advanceTimeBy( 10, TimeUnit.SECONDS );
}

输出：

Caching 1
Caching 2
Caching 3
Caching 4
Sending 1
Caching 5
Caching 6
Caching 7
Caching 8
Caching 9
Sending 2
Caching 10
. . . 

---

更新

根据您的 cmets，听起来尊重背压对您来说很重要。

假设您在某处有一个支持背压的Publisher，您可以将其转换为Flowable，如下所示：

Flowable<T> flowable = Flowable.fromPublisher( publisher );

一旦您拥有Flowable，您就可以允许多个订阅者，而不必担心每个订阅者都必须从Publisher 请求值（或者任一订阅者在建立订阅时会丢失任何事件）。你可以通过调用flowable.publish() 来创建ConnectableFlowable。

ConnectableFlowable<T> flowable = Flowable.fromPublisher( publisher ).publish();
out.send(flowable);   // calls flowable.subscribe()
cache.save(flowable); // calls flowable.subscribe()
flowable.connect();   // begins emitting values