Android Studio：RxBus联合ICompositeSubscription使用

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

• 使用 RxBus 发布变乱。
  
• 使用 ICompositeSubscription 来管理订阅。
  
• 在 Activity 中创建订阅，并确保在 Activity 烧毁时取消订阅，克制内存走漏。
  

代码实现

1、RxBus：发布和订阅变乱。

2. public class RxBus {
4.     private final SerializedSubject<Object, Object> subject;
6.     public RxBus() {
7.         subject = new SerializedSubject<>(PublishSubject.create());
8.     }
10.     public void post(Object object) {
11.         subject.onNext(object); // 发布事件
12.     }
14.     @NonNull
15.     public <T> Observable<T> toObservable(Class<T> type) {
16.         return subject.ofType(type); // 转换成指定类型的 Observable
17.     }
19.     public <T> Subscription toSubscription(Class<T> type, Action1<T> action1, Scheduler scheduler) {
20.         return this
21.                 .toObservable(type)
22.                 .subscribeOn(scheduler) // 指定线程调度器
23.                 .subscribe(action1);
24.     }
25. }

复制代码

2、 ICompositeSubscription：管理订阅。

2. public class ICompositeSubscription {
4.     private CompositeSubscription mCompositeSubscription;
6.     public ICompositeSubscription() {
7.         mCompositeSubscription = new CompositeSubscription();
8.     }
10.     public Subscription putSubscription(Subscription subscription) {
11.         mCompositeSubscription.add(subscription); // 添加订阅
12.         return subscription;
13.     }
15.     public void unSubscribe() {
16.         if (mCompositeSubscription != null) {
17.             mCompositeSubscription.clear(); // 取消所有订阅
18.         }
19.     }
20. }

复制代码

3、Event 类：界说变乱类。

2. public class Event {
4.     private String message;
6.     public Event(String message) {
7.         this.message = message;
8.     }
10.     public String getMessage() {
11.         return message;
12.     }
13. }

复制代码

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

2. public class MyActivity extends AppCompatActivity {
4.     private ICompositeSubscription subscriptionManager = new ICompositeSubscription();
5.     private RxBus mRxBus;
7.     @Override
8.     protected void onCreate(Bundle savedInstanceState) {
9.         super.onCreate(savedInstanceState);
10.         setContentView(R.layout.activity_main);
12.         mRxBus = new RxBus();  // 初始化 RxBus
14.         // 设置订阅事件的规则
15.         Subscription subscription = mRxBus.toObservable(Event.class) // 订阅 Event 类型的事件
16.                 .observeOn(AndroidSchedulers.mainThread()) // 在主线程处理
17.                 .subscribe(event -> {
18.                     Log.d("RxJava", "收到事件：" + event.getMessage());
19.                 });
21.         // 管理订阅
22.         subscriptionManager.putSubscription(subscription);
24.         // 模拟发布事件
25.         findViewById(R.id.btnSendEvent).setOnClickListener(view -> {
26.             // 发布一个 Event 类型的事件
27.             mRxBus.post(new Event("Hello from RxBus!"));
28.         });
29.     }
31.     @Override
32.     protected void onDestroy() {
33.         super.onDestroy();
34.         // 取消所有订阅，避免内存泄漏
35.         subscriptionManager.unSubscribe();
36.     }
37. }

复制代码

代码表明

•         RxBus 类：
  
  
  
  • 用于发布和订阅变乱。通过 post() 方法发布变乱，使用 toObservable() 将变乱转换为 Observable，让订阅者可以订阅该变乱。
    
         
  
•         ICompositeSubscription 类：
  
  
  
  • 用于管理多个订阅。在 onCreate() 中，订阅变乱后将订阅对象参加到 CompositeSubscription 中，确保在 onDestroy() 时可以同一取消全部订阅，克制内存走漏。
    
         
  
•         Event 类：
  
  
  
  • 简朴的变乱类，包罗一个 message 字段和构造方法，用于转达变乱数据。
    
         
  
•         MyActivity：
  
  
  
  • 在 onCreate() 中初始化 RxBus 和 ICompositeSubscription，并订阅 Event 范例的变乱。
    
  • btnSendEvent 按钮点击后，调用 RxBus.post() 发布一个变乱。
    
  • 在 onDestroy() 中调用 subscriptionManager.unSubscribe() 来取消订阅。
    
         
  

紧张概念

•         RxBus 的发布和订阅：
  
  
  
  • RxBus 是一个变乱总线，通过 toObservable() 和 post() 方法实现发布和订阅变乱。
    
  • 订阅 RxBus 后，每当有变乱发布时，订阅者会吸收到这个变乱，并做出相应。
    
         
  
•         ICompositeSubscription 的作用：
  
  
  
  • ICompositeSubscription 管理多个订阅，当 Activity 烧毁时，调用 unSubscribe() 取消全部订阅，防止内存走漏。
    
  • 在 Activity 中添加每个新的订阅并通过 putSubscription() 方法管理。