1、添加依赖
implementation "io.reactivex.rxjava3:rxjava:3.1.2"
implementation "io.reactivex.rxjava3:rxandroid:3.0.0"
2、RxBus3.kt
class RxBus3 private constructor(){
private val subjectMapper = ConcurrentHashMap<Any, MutableList<Subject<Any>>>()
private val subscriber = ConcurrentHashMap<Any, ConcurrentHashMap<Any, Observable<*>>>()
private val subscriptionMapper = ConcurrentHashMap<Any, MutableList<Disposable>>()
companion object {
val instance : RxBus3 by lazy(mode = LazyThreadSafetyMode.SYNCHRONIZED) {
RxBus3()
}
}
private fun isEmpty(collection: Collection<Subject<*>?>?): Boolean {
return null == collection || collection.isEmpty()
}
fun register(@NonNull tag: Any, @NonNull consumer: Consumer<Any>, clazz: Class<*>): Observable<Any> {
var subjectList = subjectMapper[tag]
if (null == subjectList) {
subjectList = ArrayList()
subjectMapper[tag] = subjectList
}
val subject: Subject<Any>
subject = PublishSubject.create()
subjectList.add(subject)
val sub = subject.observeOn(AndroidSchedulers.mainThread())
.subscribe(consumer) { obj: Throwable -> obj.printStackTrace() }
var subList: MutableList<Disposable>? = subscriptionMapper[clazz]
if (null == subList) {
subList = ArrayList()
}
subList.add(sub)
subscriptionMapper[clazz] = subList
var subMap: ConcurrentHashMap<Any, Observable<*>>? = subscriber[clazz]
if (subMap == null) {
subMap = ConcurrentHashMap()
}
subMap[tag] = subject
subscriber[clazz] = subMap
return subject
}
fun unregister(clazz: Class<*>) {
unsubscribe(clazz)
val sub: ConcurrentHashMap<Any, Observable<*>>? = subscriber[clazz]
if (sub != null) {
for (tag in sub.keys) {
val value = sub[tag]
if (null != value) {
unregister(tag)
}
}
}
}
private fun unsubscribe(clazz: Class<*>) {
val sub = subscriptionMapper.remove(clazz)
if (sub != null && sub.isNotEmpty()) {
val it = sub.listIterator()
while (it.hasNext()) {
val subscriptDispose = it.next()
subscriptDispose.dispose()
it.remove()
}
}
}
private fun unregister(@NonNull tag: Any) {
subjectMapper[tag]?.remove(tag)
}
fun post(@NonNull tag: Any, @NonNull content: Any) {
val subjectList = subjectMapper[tag]
if (!isEmpty(subjectList)) {
if (subjectList != null) {
for (subject in subjectList) {
subject.onNext(content)
}
}
}
}
}
3、使用
// 1、根据tag发送事件
RxBus3.instance.post("tag_any_string", "content_xxx")
RxBus3.instance.post("tag_any_object", obj)
....
//2、注册tag事件源,并指定class
RxBus3.instance.register("tag_any_object", {
obj -> run { println(obj) }
}, javaClass)
//3、删除指定class里面的所有事件
RxBus3.instance.unregister(javaClass)
非常感谢你能看到最后，如果能够帮助到你，是我的荣幸！
