RxJava 的源码解析 (六) RxJava1 Scheduler调度者概念

╰半夏微凉° 2022-05-24 01:23 116阅读 0赞

### 文章目录 ###

*   *  先了解下概念
     *   *  小案例
         *  1.0 的 Scheduler
         *  subscribeOn 调度图
         *  observeOn 调度图

## 先了解下概念 ##

### 小案例 ###

Subscription subscription = Observable.create(new Observable.OnSubscribe<String>() { 
        @Override
        public void call(Subscriber<? super String> subscriber) { 
            System.out.println(Thread.currentThread().getName());
            if (!subscriber.isUnsubscribed()) { 
                subscriber.onNext("1");
                subscriber.onNext("2");
                subscriber.onCompleted();
            }
        }
    }).subscribeOn(Schedulers.newThread())
       .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer<String>() { 
    
        @Override
        public void onCompleted() { 
            System.out.println("onCompleted");
        }
    
        @Override
        public void onError(Throwable e) { 
            e.printStackTrace();
            System.out.println("onError");
    
        }
    
        @Override
        public void onNext(String s) { 
            System.out.println(Thread.currentThread().getName());
            System.out.println("onNext：" + s);
        }
    });

### 1.0 的 Scheduler ###

*  Scheduler 是一个抽象类

<table> 
 <thead> 
  <tr> 
   <th>关键</th> 
   <th align="right">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Workerr</td> 
   <td align="right">真实做线程调度的类</td> 
  </tr> 
  <tr> 
   <td>Action0</td> 
   <td align="right">在线程中执行的操作</td> 
  </tr> 
  <tr> 
   <td>schedule</td> 
   <td align="right">实际做线程调度的方法，入参为 Action0</td> 
  </tr> 
 </tbody> 
</table>

调度这源码分析：

1.  传入不同Scheduler来使用不同的线程
2.  用Scheduler创建Worker来使用真正的线程池
3.  传入具体操作的Action0
4.  通过Schedule方法来实现调度

### subscribeOn 调度图 ###

https://www.processon.com/diagraming/5afd2fe3e4b05352c2c3f981

> 下面只是很初略的流程，最要看上面的调度图

我这里采用回溯流程 梳理

Observable对象.subscribe(new Observer<String>() { ....});

Created with Raphaël 2.2.0subscribeOperatorSubscribeOn.call()创建SubscribeOnSubscriber 对象标识<tag1>Worker.schedule(<tag1>)线程池中提交并调用 <tag1>.call()

也就解释了，subscribeOn 的线程 调度

### observeOn 调度图 ###

https://www.processon.com/view/link/5afe26b2e4b05352c2c5eb0d