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

Posted 2023-03-22

【中文标题】PublishSubject 为所有观察者调用昂贵的函数

【英文标题】：PublishSubject calls expensive function for all Observers

【发布时间】：2019-09-11 06:47:47

【问题描述】：

我有一个特定的场景，我实现了一个 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();

关于如何达到预期结果的任何想法？

【问题讨论】：

你有没有试过在链中昂贵的操作之后移动share()？

是的，似乎不影响它。 （我实际上忘记在上面的代码中添加它 - 我现在编辑它）

【参考方案1】：

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

您可以将代码更改为：

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

或者你可以改变observeResults方法来返回共享的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;

【讨论】：

这也是我怀疑发生的事情。实际共享 observable 的重要提示。谢谢。