RxJava学习笔记(二)

线程切换

• 基础概念

  在不指定线程的情况下， RxJava 遵循的是线程不变的原则，即：在哪个线程调用 subscribe()，Observable和Observer就在哪个线程，默认情况下Observable和Observer在同一个线程工作。如果需要切换线程，就需要用到 Scheduler （调度器）。

• 应用场景

  网络请求、数据库操作、遍历大列表等等

• 基础元素

  Observable.subscribeOn(Scheduler scheduler) //指定Observable发送事件的线程
  Observable.observeOn(Scheduler scheduler) //指定Observer接收事件的线程
  
  Scheduler //线程控制器，rxjava指定了以下几个常用的线程
  Schedulers.io() //io操作的线程, 通常用于网络,读写文件等io密集型的操作
  Schedulers.computation() //CPU计算密集型的操作线程, 例如需要大量计算的操作
  Schedulers.newThread() //常规的新线程
  Schedulers.newThread() //Android的主线程
  

• 代码实现

   //TODO 创建观察者
   Observer observer = ......;
   //TODO 创建被观察者
   Observable observable = .......;
   //TODO 注册和线程切换
   observable.subscribeOn(Schedulers.io()) //指定被观察者发送事件的使用io线程
   .observeOn(AndroidSchedulers.mainThread()) //指定观察者接收事件使用主线程
   .subscribe(mObserver);
  

• 特殊说明

  多次调用subscribeOn()，只有第一次有效；多次调用observeOn()，每次都有效。

• 实践案例

  在io线程里计算一个集合所有数字的和

     //生成测试数据
     ArrayList<Integer> datas = new ArrayList<>();
     for (int i = 0; i < 100000; i++) {
         datas.add(i);
     }
  
     Observable<Integer> observable = Observable.create(new Observable.OnSubscribe<Integer>() {
         @Override
         public void call(Subscriber<? super Integer> subscriber) {
             Log.d(TAG, "被观察者线程: " + Thread.currentThread().getName());
             int sum =0;
             for (Integer data : datas) {
                 sum+=data;
             }
             subscriber.onNext(sum);
         }
     });
  
     observable
             .subscribeOn(Schedulers.io())
             .observeOn(AndroidSchedulers.mainThread())
             .subscribe(new Observer<Integer>() {
                 @Override
                 public void onCompleted() {
  
                 }
  
                 @Override
                 public void onError(Throwable e) {
  
                 }
  
                 @Override
                 public void onNext(Integer integer) {
                     Log.d(TAG, "观察者线程: " + Thread.currentThread().getName());
                     Log.d(TAG, "计算完毕" +integer);
                 }
             });
  

}

​