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

问题描述

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

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

我的接口有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的项目，并且只有一个订阅者负责从发布者请求项目？

标签： javacachingrx-javareactive-streamsbackpressure

解决方案

我可能遗漏了一些东西（添加了关于我的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
. . .

更新

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

假设您有一个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

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

问题描述

解决方案

推荐阅读