安卓使用 RxJava 封装 RxBus 实现事件总线

1. 引入依赖

在 build.gradle 文件中添加 RxJava 和 RxBus 所需的依赖：

dependencies {
    implementation 'io.reactivex.rxjava2:rxjava:2.2.21'   // RxJava 依赖
    implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'  // RxAndroid 依赖（可选，用于 Android 线程切换）
}

2. 定义 RxBus 类

创建一个封装了 RxJava 的 RxBus 类，它负责管理消息的发布与订阅。

public class RxBus {

    private final PublishSubject<Object> bus;

    private RxBus() {
        bus = PublishSubject.create();
    }

    private static class Holder {
        private static final RxBus INSTANCE = new RxBus();
    }

    public static RxBus getInstance() {
        return Holder.INSTANCE;
    }

    // 发送事件
    public void post(Object event) {
        bus.onNext(event);
    }

    // 订阅事件
    public <T> Observable<T> toObservable(Class<T> eventType) {
        return bus.ofType(eventType);
    }
}

• PublishSubject 是 RxJava 提供的一个 Subject，它可以同时作为观察者和被观察者。当你调用 onNext 时，所有订阅者都会接收到消息。
• toObservable 方法允许我们根据事件类型过滤消息。

3. 创建自定义事件类

根据不同的需求定义不同的事件类。例如，创建一个表示登录成功的事件：

public class LoginEvent {
    private String userName;
    private String userId;

    public LoginEvent(String userName, String userId) {
        this.userName = userName;
        this.userId = userId;
    }

    public String getUserName() {
        return userName;
    }

    public String getUserId() {
        return userId;
    }
}

4. 发送事件

当需要发送事件时，可以通过 RxBus.getInstance().post() 方法将事件发布出去：

// 发送登录成功事件
LoginEvent loginEvent = new LoginEvent("JohnDoe", "12345");
RxBus.getInstance().post(loginEvent);

5. 在多个地方监听事件

在不同的地方（如 Activity 或 Fragment），你可以通过 RxBus 监听事件。为了避免内存泄露，我们使用 CompositeDisposable 来管理订阅。

public class MainActivity extends AppCompatActivity {

    private CompositeDisposable compositeDisposable = new CompositeDisposable();

    @Override
    protected void onStart() {
        super.onStart();

        // 订阅事件
        Disposable disposable = RxBus.getInstance()
            .toObservable(LoginEvent.class)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(loginEvent -> {
                // 处理登录事件
                Log.d("MainActivity", "User Logged In: " + loginEvent.getUserName());
            });

        compositeDisposable.add(disposable); // 添加到 CompositeDisposable 管理
    }

    @Override
    protected void onStop() {
        super.onStop();
        compositeDisposable.clear(); // 清除所有订阅，避免内存泄露
    }
}

• compositeDisposable.add(disposable) 用于添加订阅到 CompositeDisposable 中。
• 在 onStop() 中调用 compositeDisposable.clear() 来清除所有的订阅，避免因为 Activity 或 Fragment 销毁后仍然存在未取消的订阅导致内存泄露。