Android Studio：RxBus结合ICompositeSubscription使用

        我现在想用 RxBus 来发布和订阅事件，同时使用 ICompositeSubscription 来管理订阅。跟前一个博客的区别在于，事件流的产生方式不同，更加得全面。

目标

• 使用 RxBus 发布事件。
• 使用 ICompositeSubscription 来管理订阅。
• 在 Activity 中创建订阅，并确保在 Activity 销毁时取消订阅，避免内存泄漏。

代码实现

1、RxBus：发布和订阅事件。

public class RxBus {private final SerializedSubject<Object, Object> subject;public RxBus() {subject = new SerializedSubject<>(PublishSubject.create());}public void post(Object object) {subject.onNext(object); // 发布事件}@NonNullpublic <T> Observable<T> toObservable(Class<T> type) {return subject.ofType(type); // 转换成指定类型的 Observable}public <T> Subscription toSubscription(Class<T> type, Action1<T> action1, Scheduler scheduler) {return this.toObservable(type).subscribeOn(scheduler) // 指定线程调度器.subscribe(action1);}
}

2、 ICompositeSubscription：管理订阅。

public class ICompositeSubscription {private CompositeSubscription mCompositeSubscription;public ICompositeSubscription() {mCompositeSubscription = new CompositeSubscription();}public Subscription putSubscription(Subscription subscription) {mCompositeSubscription.add(subscription); // 添加订阅return subscription;}public void unSubscribe() {if (mCompositeSubscription != null) {mCompositeSubscription.clear(); // 取消所有订阅}}
}

3、Event 类：定义事件类。

public class Event {private String message;public Event(String message) {this.message = message;}public String getMessage() {return message;}
}

4、Activity 代码：在 Activity 中，使用 RxBus 发布和订阅事件，并使用 ICompositeSubscription 来管理订阅。

public class MyActivity extends AppCompatActivity {private ICompositeSubscription subscriptionManager = new ICompositeSubscription();private RxBus mRxBus;@Overrideprotected void onCreate(Bundle savedInstanceState) {super.onCreate(savedInstanceState);setContentView(R.layout.activity_main);mRxBus = new RxBus();  // 初始化 RxBus// 设置订阅事件的规则Subscription subscription = mRxBus.toObservable(Event.class) // 订阅 Event 类型的事件.observeOn(AndroidSchedulers.mainThread()) // 在主线程处理.subscribe(event -> {Log.d("RxJava", "收到事件：" + event.getMessage());});// 管理订阅subscriptionManager.putSubscription(subscription);// 模拟发布事件findViewById(R.id.btnSendEvent).setOnClickListener(view -> {// 发布一个 Event 类型的事件mRxBus.post(new Event("Hello from RxBus!"));});}@Overrideprotected void onDestroy() {super.onDestroy();// 取消所有订阅，避免内存泄漏subscriptionManager.unSubscribe();}
}

代码解释

1. RxBus 类：

   • 用于发布和订阅事件。通过 post() 方法发布事件，使用 toObservable() 将事件转换为 Observable，让订阅者可以订阅该事件。

2. ICompositeSubscription 类：

   • 用于管理多个订阅。在 onCreate() 中，订阅事件后将订阅对象加入到 CompositeSubscription 中，确保在 onDestroy() 时可以统一取消所有订阅，避免内存泄漏。

3. Event 类：

   • 简单的事件类，包含一个 message 字段和构造方法，用于传递事件数据。

4. MyActivity：

   • 在 onCreate() 中初始化 RxBus 和 ICompositeSubscription，并订阅 Event 类型的事件。
   • btnSendEvent 按钮点击后，调用 RxBus.post() 发布一个事件。
   • 在 onDestroy() 中调用 subscriptionManager.unSubscribe() 来取消订阅。

重要概念

• RxBus 的发布和订阅：

  • RxBus 是一个事件总线，通过 toObservable() 和 post() 方法实现发布和订阅事件。
  • 订阅 RxBus 后，每当有事件发布时，订阅者会接收到这个事件，并做出响应。

• ICompositeSubscription 的作用：

  • ICompositeSubscription 管理多个订阅，当 Activity 销毁时，调用 unSubscribe() 取消所有订阅，防止内存泄漏。
  • 在 Activity 中添加每个新的订阅并通过 putSubscription() 方法管理。