RxBus 1

代码

object RxBus {

    private val subject: Subject<Any> = PublishSubject.create<Any>().toSerialized()

    fun post(any: Any) {
        subject.onNext(any)
    }

    fun <T> registerEvent(clazz: Class<T>, consumer: Consumer<T>) {
        subject.ofType(clazz).subscribe(consumer)
    }

}

使用

class SecondActivity : AppCompatActivity() {

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_second)
        
        RxBus.registerEvent(String::class.java, Consumer {
            textView.text = it
        })

        textView.setOnClickListener {
            RxBus.post("hello world")
        }
    }

}

原理分析

  subject.ofType(clazz).subscribe(consumer)

ofType 源码：

    public final <U> Observable<U> ofType(final Class<U> clazz) {
        ObjectHelper.requireNonNull(clazz, "clazz is null");
        return filter(Functions.isInstanceOf(clazz)).cast(clazz);
    }

根据 简单操作符链式调用的实质
S.F.C -> A 等价于，
S -> F(C(A))，F 代表 FilterObserver。
S -> F 代表，把 F 加入 subject 的通知集合。
根据 Subject，每当

    fun post(any: Any) {
        subject.onNext(any)
    }

一个 any 就会被发射给 F(C(A))，F 执行过滤，调用 C.onNext，
C 执行类型转换，最终调用 A.accept。