RxJava超简单实现EventBus基本功能

目录

目录

前言

身为Android开发大家都应该熟悉EventBus这款开源库，但是有时候我们往往只需要EventBus的基本功能不需要其他多余的功能怎么办呢？如果这时你的项目刚好使用的RxJava那么你就有福了，完全可以通过极少的代码来实现EventBus的基本功能。

效果展示

演示的主要是在第二个界面发送事件给第一个界面然后并展示出来

实现原理

这里使用的是RxJava里的Subject，我们从它的源码可以看出它既是Observable同时也是Observer，正好可以作为两者之间沟通的桥梁，同时这里也使用了PublishSubject.create<Any>().toSerialized()来生成Subject这样可以保证线程安全。

实现步骤

1.编写RxBus类
这里使用Kotlin进行编写可以极大简化代码

object RxBus {
    private var mBus:Subject<Any> = PublishSubject.create<Any>().toSerialized()//保证线程安全
    fun post(any: Any){
        mBus.onNext(any)
    }
    /**
     * 过滤信息，此方法生成的Observable只对指定类型进行监听
     */
    fun <T> toObservable(clazz:Class<T>):Observable<T>{
        return mBus.ofType(clazz)
    }
    /**
     * 不过滤信息,监听任何类型的消息
     */
    fun toObservable():Observable<Any>{
        return mBus
    }
}

2.编写RxEvent类（用于事件传递）

class RxEvent(private var msg: String) {
    fun getMsg():String{
        return msg
    }
}

3.在第一个Activity中进行监听

class MainActivity : AppCompatActivity() {
    private lateinit var disposable:Disposable
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)
        //进行监听
        disposable = RxBus.toObservable(RxEvent::class.java)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe({
                    main_tv.text = it.getMsg()
                })
        main_bt.setOnClickListener {
            startActivity(Intent(this@MainActivity,SecondActivity::class.java))
        }
    }

    override fun onDestroy() {
        super.onDestroy()
        //取消监听
        if(!disposable.isDisposed){
            disposable.dispose()
        }
    }
}

4.在第二个Activity中进行事件发送

class SecondActivity:AppCompatActivity (){
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_second)
        second_bt.setOnClickListener {
            RxBus.post(RxEvent("我是第二界面的信息"))
        }
    }
}

两个Activity的布局文件如下：

• activity_main.xml：

<?xml version="1.0" encoding="utf-8"?>
<LinearLayout xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:app="http://schemas.android.com/apk/res-auto"
    xmlns:tools="http://schemas.android.com/tools"
    android:orientation="vertical"
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    tools:context="com.itfitness.rxbusdemo.MainActivity">
    <Button
        android:id="@+id/main_bt"
        android:text="第二界面"
        android:layout_width="match_parent"
        android:layout_height="wrap_content" />
    <TextView
        android:id="@+id/main_tv"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        />

</LinearLayout>

• activity_second.xml：

<?xml version="1.0" encoding="utf-8"?>
<LinearLayout
    xmlns:android="http://schemas.android.com/apk/res/android"
    android:orientation="vertical"
    android:layout_width="match_parent"
    android:layout_height="match_parent">
    <Button
        android:id="@+id/second_bt"
        android:text="发送"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content" />
</LinearLayout>

拓展：使RxBus支持背压

由于在RxJava2.x中Subject已经不支持背压了，因此如果想支持背压可以将Subject换成PublishProcessor。

object RxBus {
    private var mBus:FlowableProcessor<Any> = PublishProcessor.create<Any>().toSerialized()//保证线程安全
    fun post(any: Any){
        mBus.onNext(any)
    }
    /**
     * 过滤信息，此方法生成的Observable只对指定类型进行监听
     */
    fun <T> toObservable(clazz:Class<T>):Flowable<T>{
        return mBus.ofType(clazz)
    }
    /**
     * 不过滤信息,监听任何类型的消息
     */
    fun toObservable():Flowable<Any>{
        return mBus
    }
}