PublishSubject 为所有观察者调用昂贵的函数

Question

我有一个特定的场景，我实现了一个 PublishSubject 以根据自定义事件发出项目。对于将发出的每个项目，我还需要保留该值（一项昂贵的操作）。我想要实现的是拥有一个函数（如 map），它将为所有观察者调用一次，然后每个观察者通过 onNext() 方法接收项目。

主题：

static final PublishSubject<SomeResult> commonSubject = PublishSubject.create()

触发器（发射物品）：

commonSubject.onNext(new SomeResult())

暴露主题（将被控制器使用）：

public static Observable<SomeResult> observeResults() {
    return commonSubject.share();
}

控制器：

public Observable<SomeResult> observeResults() {
    return CustomConsumer.observeResults()
            .observeOn(Schedulers.single());
}

订阅者：

CustomControllerResult.observeResults().subscribe(result -> doSomething());
CustomControllerResult.observeResults().subscribe(result -> doSomethingElse());

每个观察者都按预期接收项目，但如果我将昂贵的操作添加到控制器，将为每个观察者调用（我不想要的东西）：

public Observable<SomeResult> observeResults() {
    return CustomConsumer.observeResults()
            .observeOn(Schedulers.single())
            .compose(persistResult())
            .compose(logResult())
            .share();
}

关于如何实现预期结果的任何想法？

Answer 1

问题是每次调用 observeResults() 时，它都会使用 share 运算符创建一个新的 Observable。但是创建的 Observable 没有与订阅者共享。

您可以将代码更改为：

Observable<SomeResult> observable = CustomControllerResult.observeResults()
observable.subscribe(result -> doSomething());
observable.subscribe(result -> doSomethingElse());

或者您可以将 observeResults 方法更改为 return 共享 Observable：

static final PublishSubject<SomeResult> commonSubject = PublishSubject.create()

static final Observable<SomeResult> observable = commonSubject
    .observeOn(Schedulers.single())
    .compose(persistResult())
    .compose(logResult())
    .share();

public static Observable<SomeResult> observeResults() {
    return observable;
}

PublishSubject 为所有观察者调用昂贵的函数

PublishSubject calls expensive function for all Observers

android

rx-java2

publishsubject