博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
RxJava操作符之Share, Publish, Refcount
阅读量:7079 次
发布时间:2019-06-28

本文共 3261 字,大约阅读时间需要 10 分钟。

看源码知道.share()操作符是.publish().refcount()调用链的包装。

先来看ConnectedObservable

“ConnectedObservable” – This is a kind of observable which doesn’t emit items even if subscribed to.It only starts emitting items after its .connect() method is called.

因为这个原因,在ConnectedObservable的connect这个方法被调用之前,connected obesrvable也被认为是“冷”和不活跃。

再看publish方法

.publish()– This method allows us to change an ordinary observable into a “ConnectedObservable”.Simply call this method on an ordinary observable and it becomes a connected one.

现在我们知道了share操作符的1/2,那么为什么需要运用Connected Observable这个操作符呢?文档上是这么写的:

In this way you can wait for all intended Subscribers to subscribe to the Observable before the Observable begins emitting items.

这就意味着publish可以调用多个subscriber。当你有超过一个订阅者的时候,处理每个订阅和正确的销毁他们变得棘手。 为了使这个更方便,Rx发明了这个魔法操作符refcount():

refcount() – This operator keeps track of how many subscribers are subscribed to the resulting Observable andrefrains from disconnecting from the source ConnectedObservable until all such Observables are unsubscribed.

refcount本质上在后台维护着一个引用计数器,当一个subscription需要取消订阅或者销毁的时候,发出一个正确的动作。

我们再次看一下debouncedBuffer的例子,看一下在哪,share是怎么用的。

Observable tapEventEmitter = _rxBus.toObserverable().share();// which is really the same as:Observable tapEventEmitter = _rxBus.toObserverable().publish().refcount();

我们现在有了一个"shareable"的名字叫"tapEventEmitter"的observable。 因为他是可以分享的,而且还不是“live”(share操作符中的publish调用使其变成一个ConnectedObservable), 我们可以用他构成我们的Observables,而且要确保我们有了一个原始的observable的引用 (这个例子中原始的observable是_rxBus.toObserverable()).

Observable tapEventEmitter = _rxBus.toObserverable().share();Observable debouncedEventEmitter = tapEventEmitter.debounce(1, TimeUnit.SECONDS);tapEventEmitter.buffer(debouncedEventEmitter)//...

所有的这一切看起来都很好。然而这个实现会有一个可能的竞争条件。因为这有两个subscribers(debounce and buffer)而且会在不同的时间点发生,所以竞争条件就会发生。 记住RxBus是由hot/live Subject支持的不断的发射数据。通过使用share操作符,我们保证引用的是同一个资源。 而不是subscribers在不同的时间点订阅,他们会收到准确的相同的数据。

The race condition is when the two consumers subscribe. Often on a hot stream it doesn’t matter when subscribers come and go,and refCount is perfect for that.The race condition refCount protects against is having only 1 active subscription upstream. However,if 2 Subscribers subscribe to a refcounted stream that emits 1, 2, 3, 4, 5, the first may get 1, 2, 3, 4, 5 and the second may get 2, 3, 4, 5.To ensure all subscribers start at exactly the same time and get the exact same values, refCount can not be used.Either ConnectableObservable with a manual, imperative invocation of connect needs to be done, or the variant of publish(function)which connects everything within the function before connecting the upstream.

在我们的用法中几乎立即执行所以没有什么关系。但是我们原始的意图是把debouncedBuffer方法作为一个单独的操作符。 如果相同的事件没有被发射出去,从概念上看起来是不正确的。

通过Bean后来的建议,我添加了一个更好的第三方的实现,用来处理这种竞争条件。

// don't start emitting items just yet by turning the observable to a connected oneConnectableObservable tapEventEmitter = _rxBus.toObserverable().publish();tapEventEmitter.publish((Func1) (stream) -> {// inside `publish`, "stream" is truly multicasted // applying the same technique for getting a debounced buffer sequence return stream.buffer(stream.debounce(1, TimeUnit.SECONDS)); }).subscribe((Action1) (taps) { _showTapCount(taps.size()); }); // start listening to events now tapEventEmitter.connect();

转载地址:http://egdml.baihongyu.com/

你可能感兴趣的文章
阿里巴巴收购以色列VR公司,大厂死磕VR为哪般?
查看>>
使用Xamarin实现跨平台移动应用开发
查看>>
webpack-dev-server 热替换配置详解
查看>>
随机森林算法4种实现方法对比测试:DolphinDB速度最快,XGBoost表现最差
查看>>
架构设计复杂度的6个来源
查看>>
如何成功地在亚洲植入敏捷和DevOps
查看>>
银行建中台跟阿里建中台有什么不同?
查看>>
实现AGI还要多久?Hinton与AlphaGo之父这样回答
查看>>
Atlassian的Stash数据中心为Git提供了高可用性及可伸缩性
查看>>
Adaptive Execution让Spark SQL更高效更好用
查看>>
Swift 烧脑体操(五)- Monad
查看>>
中国在两年内赶超美国AI?李开复:不一定
查看>>
OpsRamp推出AIOps推理引擎
查看>>
C#未来新特性:静态委托和函数指针
查看>>
性能之巅:Linux网络性能分析工具
查看>>
Facebook曝至今最严重安全漏洞,超过5000万用户受影响
查看>>
简单介绍我的开源小工具:SanicDB
查看>>
我做SAP CRM One Order redesign的一些心得体会
查看>>
第二十二章:动画(十)
查看>>
个推微服务网关架构实践
查看>>