美文网首页
Android搭建应用框架系列之RxBus

Android搭建应用框架系列之RxBus

作者: Goach | 来源:发表于2017-11-16 11:48 被阅读283次

    前言

    项目中,往往少不了事件总线,以前一般使用EventBus,但现在由于Rx系列的强大,也就投入了RxBus的怀抱。下面就来说说RxBus

    背压策略

    RxJava1.x版本的时候,可能会遇到MissingBackpressureExceptionOOM这样的问题,这是很多事件不能被正常的背压处理的导致的,在RxJava2.x版本引入了Flowable来处理背压的问题,那么,什么是背压呢?

    背压策略概念:
    RxJava在异步事件处理并且被观察者处理事件的速度比观察者处理事件的速度快的情况下,通知被观察者降低发送速度的策略

    主要的实现原理就是:
    Observables分为两种主要类型:Hot ObservablesCold ObservablesCold Observables是有订阅事件的观察者才会开始发送事件,比如from,create,just,timer,interval,range,zip等操作符;而Hot Observables不管是否有订阅者,创建完后就直接发送事件。所以Hot Observable是不支持背压式,而Cold Observable是部分操作符支持背压式策略的。而不支持背压式的操作符可以通过sample( ) 、 throttleLast( )、 throttleFirst( )、 throttleWithTimeout( ) 、 debounce( )等等这些节流操作符去控制观察者的发送事件速度。关于这方面的知识,这篇文章关于RxJava最友好的文章——背压(Backpressure)就写的很不错。

    发送事件

    实现步骤

    • 创建一个FlowableProcessor(Any)
    • post(o:Any)方法,发送事件函数, 同时传入需要发送的信息
    • toFlowable()返回FlowableProcessor<Any>对象

    最后得到的代码如下:

    object RxBus{
        private val bus:FlowableProcessor<Any> = PublishProcessor.create<Any>().toSerialized()
        fun post(t:Any){
            bus.onNext(t)
        }
        fun toFlowable(): FlowableProcessor<Any> {
            return bus
        }
    }
    

    接收事件

    实现步骤

    • 通过toFlowable()获取到FlowableProcessor<Any>对象,
    • 传入需要接收的信息类型,通过ofType(classType)获取
    • subscribe订阅,获取到信息后进一步的操作。

    最后得到的代码如下:

    class Bus(val sub: CompositeDisposable){
        fun <T : Event> subscribe(classType: Class<T>, onNext:((res: T) ->Unit)?=null,
                                  onError:((e: Throwable) ->Unit)?=null,
                                  onComplete:(() ->Unit)?=null) {
            val subscribe = RxBus.toFlowable()
                    .ofType(classType)
                    .subscribe({
                        if(onNext!=null) onNext(it)
                    },{
                        if(onError!=null) onError(it)
                    },{
                        if(onComplete!=null) onComplete()
                    })
            sub.add(subscribe)
        }
    }
    
    fun BaseActivity.bus(init: Bus.() -> Unit): Bus {
        val bus = Bus(sub)
        bus.init()
        return bus
    }
    
    fun BaseFragment.bus(init: Bus.() -> Unit): Bus {
        val bus = Bus(sub)
        bus.init()
        return bus
    }
    

    使用

    定义一个Event

    open class Event
    class LogoutEvent : Event()
    

    然后发送信息

    RxBus.post(LogoutEvent())
    

    最后接收信息

     bus {
                subscribe(LogoutEvent::class.java,{
                    //进行退出登录操作
                })
            }
    

    在获取信息的时候传入的CompositeDisposableBaseActivity/BaseFragmentCompositeDisposable,所以需要在BaseActivity/BaseFragmentonDestroy()里面进行clear

     override fun onDestroy() {
            sub.clear()
            super.onDestroy()
        }
    

    这样,一个简单的RxBus就封装好了。

    相关文章

      网友评论

          本文标题:Android搭建应用框架系列之RxBus

          本文链接:https://www.haomeiwen.com/subject/oexipxtx.html