• 7. RxJava总结

    1.Rx思维 (起点到终点链条不断,响应式编程)

    起点(分发事件(Path): 我饿了)--下楼--去餐厅---点餐--->终点(吃饭消费事件)

    程序中的例子

    起点(分发事件,点击登录)---登录API---请求服务器---获取响应码---->终点(更新UI登录成功 消费事件)

    2.核心思想

    有一个起点 和 一个终点, 起点开始流向我们的"事件" 把事件流向到终点,只不过在流向的过程中,可以增加拦截,

    拦截时可以对"事件进行改变", 终点只关心它上一个拦截;

    ---------------------------------------------------------------

    Observable 根据上层变化而变化过滤 Observer

    被观察者 订阅(上层区域 和 下层区域 关联) 观察者

    起点 终点

    ---------------------------------------------------------------

    标准观察者: RxJava的观察者模式:

    3.RxJava配合Retrofit

    1> 数据模块

    1. // 总数据Bean
    2. public class ProjectBean {
    3.     private int errorCode;
    4.     private String errorMsg;
    5.     private List data;
    6.  
    7.  
    8.     public static class DataBean {
    9.         private int courseId;
    10.         private int id;
    11.         private String name;
    12.         private int order;
    13.         private int parentChapterId;
    14.         private boolean userControlSetTop;
    15.         private int visible;
    16.         private List> children;
    17.     }
    18. }
    19.  
    20.  
    21. // Item数据
    22. public class ProjectItem {
    23.     private DataBean data;
    24.     private int errorCode;
    25.     private String errorMsg;
    26.     public static class DataBean {
    27.         private int curPage;
    28.         private int offset;
    29.         private boolean over;
    30.         private int pageCount;
    31.         private int size;
    32.         private int total;
    33.         private List datas;
    34.         public static class DatasBean {
    35.  
    36.  
    37.             private String apkLink;
    38.             private String desc;
    39.             private String envelopePic;
    40.             private boolean fresh;
    41.             private int id;
    42.             private String link;
    43.             private String niceDate;
    44.             private String origin;
    45.             private String prefix;
    46.             private String projectLink;
    47.             private List tags;
    48.  
    49.  
    50.             public static class TagsBean {
    51.                 private String name; // 项目
    52.                 private String url; ///project/list/1?cid=294
    53.             }
    54.         }
    55.     }
    56. }

    2> Api模块

    1. public interface WanAndroidApi {
    2.     //总数据
    3.     @GET("project/tree/json")
    4.     Observable getProject(); //异步线程耗时操作
    5.  
    6.  
    7.     //Item数据
    8.     @GET("project/list/{pageIndex}/json") //?cid=294 异步线程 耗时操作
    9.     Observable getProjectItem(@Path("pageIndex") int pageIndex, @Query("cid") int cid);
    10. }
    11.  
    12.

    3> 业务操作

    


    1. public class UseActivity extends AppCompatActivity {
    2.     private WanAndroidApi api;
    3.     public static final String TAG = "rxJava";
    4.     @Override
    5.     protected void onCreate(Bundle savedInstanceState) {
    6.         super.onCreate(savedInstanceState);
    7.         setContentView(R.layout.activity_use);
    8.         api = HttpUtil.getOnlineCookieRetrofit().create(WanAndroidApi.class);
    9.         antiShakeAction();
    10.     }
    11.  
    12.  
    13. //TODO Retrofit+RxJava 查询 项目分类 (总数据查询)
    14. Disposable disposable;
    15. public void getProjectAction(View view) {
    16.     disposable = api.getProject()
    17.             .subscribeOn(Schedulers.io()) //给上面分配异步线程
    18.             .observeOn(AndroidSchedulers.mainThread())//给下面主线程
    19.             .subscribe(new Consumer<ProjectBean>() {//简化版
    20.                 @Override
    21.                 public void accept(ProjectBean projectBean) throws Exception {
    22.                     Log.d(TAG, "accept: " + projectBean.toString());//可以做UI操作
    23.                 }
    24.             });
    25. }
    26.  
    27.  
    28. //TODO Retrofit+RxJava 查询 项目分类的 去获取项目列表数据
    29. public void getProjectListAction(View view) {
    30.     disposable = api.getProjectItem(1, 294)
    31.             .compose(rxUD())
    32.             .subscribe(projectItem -> {
    33.                 Log.d(TAG, "accept: " + projectItem.toString());//可以做UI操作
    34.             });
    35. }
    36.  
    37.  
    38. /**
    39.  * 封装我们线程调度的操作
    40.  * UD : upstream 上游 , downstream 下游
    41.  */
    42. public static <UD> ObservableTransformer<UD, UD> rxUD() {
    43.     return upstream -> {
    44.         return upstream.subscribeOn(Schedulers.io())  //给上面代码(subscribeOn)分配异步线程
    45.                 .observeOn(AndroidSchedulers.mainThread()); //给下面代码(observeOn)分配主线程
    46.     };
    47. }
    48.  
    49.  
    50. @Override
    51. protected void onDestroy() {
    52.     super.onDestroy();
    53.     if (disposable != null && !disposable.isDisposed()) {
    54.         disposable.dispose();
    55.     }
    56. }
    57.  
    58.

    4.防抖 rxBinding 与 网络嵌套

    1. public class UseActivity extends AppCompatActivity {
    2.     private WanAndroidApi api;
    3.     public static final String TAG = "rxJava";
    4.  
    5.  
    6.     @Override
    7.     protected void onCreate(Bundle savedInstanceState) {
    8.         super.onCreate(savedInstanceState);
    9.         setContentView(R.layout.activity_use);
    10.         api = HttpUtil.getOnlineCookieRetrofit().create(WanAndroidApi.class);
    11.         antiShakeAction();
    12.     }
    13.  
    14.  
    15.  
    16.  
    17.     //TODO Retrofit+RxJava 查询 项目分类 (总数据查询)
    18.     Disposable disposable;
    19.  
    20.  
    21.     /**
    22.      * TODO 自定义 功能防抖
    23.      */
    24.     private void antiShakeAction() {
    25.         //对Button防抖动
    26.         Button btn = findViewById(R.id.bt_anti_shake);
    27.         disposable = RxView.clicks(btn)
    28.                 .throttleFirst(2000, TimeUnit.MILLISECONDS) //两秒内 响应一次
    29.                 .subscribe(new Consumer<Object>() {
    30.                     @Override
    31.                     public void accept(Object o) throws Exception {
    32.                         Log.d(TAG, "accept: 我响应了一次");
    33.                     }
    34.                 });
    35.     }
    36.  
    37.  
    38.     //TODO 防抖功能 + 网络嵌套 (解决网络嵌套问题) flatMap
    39.     private void antiShakeActionUpdate() {
    40.         //对Button防抖动
    41.         Button btn = findViewById(R.id.bt_anti_shake);
    42.         disposable = RxView.clicks(btn)//使用RxView来进行防抖
    43.                 .throttleFirst(2000, TimeUnit.MILLISECONDS) //两秒内 响应一次
    44.                 //我只给下面切换 异步线程
    45.                 .observeOn(Schedulers.io())
    46.                 //使用flatMap 自己分发 10个数据  onNext(1)-> 给下面 1-->多分发 10个数据
    47.                 //将Object 转换成 ObservableSource 传递给下一层
    48.                 .flatMap(new Function<Object, ObservableSource<ProjectBean>>() {
    49.                     @Override
    50.                     public ObservableSource<ProjectBean> apply(Object o) throws Exception {
    51.                         return api.getProject(); //返回 Observable
    52.                     }
    53.                 })
    54.                 //将 ProjectBean 转换为 ObservableSource 传递给下一层
    55.                 .flatMap(new Function<ProjectBean, ObservableSource<ProjectBean.DataBean>>() {
    56.                     @Override
    57.                     public ObservableSource<ProjectBean.DataBean> apply(@NonNull ProjectBean projectBean) {
    58.                         return Observable.fromIterable(projectBean.getData());//自己搞一个发射器 发多次;;
    59.                     }
    60.                 })
    61.                 // 将ProjectBean.DataBean 转换成 ObservableSource 传递给下一层
    62.                 .flatMap(new Function<ProjectBean.DataBean, ObservableSource<ProjectItem>>() {
    63.                     @Override
    64.                     public ObservableSource<ProjectItem> apply(@NonNull ProjectBean.DataBean dataBean) {
    65.                         return api.getProjectItem(1, dataBean.getId()); //获取Item数据
    66.                     }
    67.                 })
    68.                 .observeOn(AndroidSchedulers.mainThread()) //给下面使用的线程
    69.                 .subscribe(new Consumer<ProjectItem>() {
    70.                     @Override
    71.                     public void accept(ProjectItem projectItem) {
    72.                         //如果我要更新UI,要切会主线程
    73.                         Log.d(TAG, "antiShakeAction: " + projectItem.toString());
    74.                     }
    75.                 });
    76.     }
    77.  
    78.  
    79.     @Override
    80.     protected void onDestroy() {
    81.         super.onDestroy();
    82.         if (disposable != null && !disposable.isDisposed()) {
    83.             disposable.dispose();
    84.         }
    85.     }

    5.doOnNext(频繁的线程切换)

    请求服务器注册(耗时操作)

    ↓

    更新注册UI(main线程)

    ↓

    请求服务器登录(耗时操作)

    ↓

    更新注册UI(main线程)

    GitHub - ReactiveX/RxJava: RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.

    ReactiveX

    玩Android 开放API-玩Android - wanandroid.com

    https://www.bejson.com/json2javapojo/new/%27

    玩Android 开放API-玩Android - wanandroid.com

    https://www.bejson.com/json2javapojo/new/%27

    1> RxJava Hook

    1. RxJava 1.x
    2. @NonNull
    3. public static  Observable onAssembly(@NonNull Observable source) {
    4.     //预留给2.x
    5.     return source;
    6. }
    7.  
    8.  
    9. RxJava2.x  RxJavaPlugins中
    10. @NonNull
    11. public static  Observable onAssembly(@NonNull Observable source) {
    12.     //默认情况下 f==null
    13.     Functionsuper Observable, ? extends Observable> f = onObservableAssembly;
    14.     if (f != null) {
    15.         return apply(f, source);
    16.     }
    17.     return source;
    18. }
    19. public static void setOnObservableAssembly(@Nullable Functionsuper Observable, ? extends Observable> onObservableAssembly) {
    20.      if (lockdown) throw new IllegalStateException("Plugins can't be changed anymore");
    21.  
    22.  
    23.      RxJavaPlugins.onObservableAssembly = onObservableAssembly;
    24. }
    25. public static void main(String[] args) {
    26.      //Hook之前的监听 是static 全局的 Hook 很多操作符都会经过[RxJavaPlugins.onAssembly]
    27.      RxJavaPlugins.setOnObservableAssembly(new Function<Observable, Observable>() {
    28.          @Override
    29.          public Observable apply(Observable observable) throws Exception {
    30.              System.out.println("apply: 整个项目全局监听到底有多少地方使用了 RxJava"+observable);
    31.              return observable;//不破坏人家的功能
    32.          }
    33.      });
    34.     testJust();
    35. }
    36. public static void testJust(){
    37.     Observable.just(1, 2, 4).subscribe(new Consumer<Integer>() {
    38.         @Override
    39.         public void accept(Integer integer) throws Exception {
    40.             System.out.println("integer==>" + integer);
    41.         }
    42.     });
    43. }
    44. //执行结果
    45. 整个项目全局监听 observable: io.reactivex.internal.operators.observable.ObservableFromArray@64a294a6
    46. integer==>1
    47. integer==>2
    48. integer==>4

    2> RxJava的观察者模式

    1. //起点被观察者
    2. Observable.just(PATH) //TODO 第二步 内部会分发 返回 Observable
    3.         //TODO 第三步 卡片式拦截  将String 转换为 Bitmap
    4.         .map(new Function() {
    5.             @Override
    6.             public Bitmap apply(String s) throws Exception {
    7.                 URL url = new URL(PATH);
    8.                 HttpURLConnection conn = (HttpURLConnection) url.openConnection();
    9.                 conn.setConnectTimeout(5000);
    10.                 int responseCode = conn.getResponseCode();
    11.                 if (responseCode == HttpURLConnection.HTTP_OK) {
    12.                     InputStream inputStream = conn.getInputStream();
    13.                     return BitmapFactory.decodeStream(inputStream);
    14.                 }
    15.                 return null;
    16.             }
    17.         })
    18.         //日志记录
    19.         .map(new Function() {
    20.             @Override
    21.             public Bitmap apply(Bitmap bitmap) throws Exception {
    22.                 Log.d(TAG, "apply: " + getNormalTime(System.currentTimeMillis()) + "下载了图片");
    23.                 return bitmap;//将上个返回的数据继续流给下游;
    24.             }
    25.         })
    26.         .map(new Function() {
    27.             @Override
    28.             public Bitmap apply(Bitmap bitmap) throws Exception {
    29.                 Paint paint = new Paint();
    30.                 paint.setTextSize(88);
    31.                 paint.setColor(Color.RED);
    32.                 return drawTextToBitmap(bitmap, "NorthStar", paint, 88, 88);
    33.             }
    34.         })
    35.         //TODO 将上下游调度网络的操作放入compose中
    36.         //.subscribeOn(Schedulers.io()) //给上面代码 分配异步线程
    37.         //.observeOn(AndroidSchedulers.mainThread()) //给下面代码分配主线程
    38.         .compose(rxUD())
    39.         //TODO 订阅 起点 和 终点 订阅起来  上层改变 下层会响应改变
    40.         .subscribe(
    41.                 //终点观察者
    42.                 new Observer() {
    43.                     @Override//TODO 第一步 订阅开始 预备开始要分发
    44.                     public void onSubscribe(Disposable d) {
    45.                         progressDialog = new ProgressDialog(DownloadActivity.this);
    46.                         progressDialog.setTitle("下载图片中...");
    47.                         progressDialog.show();
    48.                     }
    49.  
    50.  
    51.                     @Override// TODO 第四步 拿到事件
    52.                     public void onNext(Bitmap bitmap) {
    53.                         iv.setImageBitmap(bitmap);
    54.                     }
    55.  
    56.  
    57.                     @Override//错误事件
    58.                     public void onError(Throwable e) {
    59.                         Log.d(TAG, "onError: " + e.getMessage());
    60.                     }
    61.  
    62.  
    63.                     @Override//TODO 第五步  完成事件
    64.                     public void onComplete() {
    65.                         if (progressDialog != null) {
    66.                             progressDialog.dismiss();
    67.                         }
    68.                     }
    69.                 });
    70.  
    71.  
    72.     /**
    73.      * 封装我们线程调度的操作
    74.      * UD : upstream 上游 , downstream 下游
    75.      */
    76.     public static  ObservableTransformer rxUD() {
    77.         return new ObservableTransformer() {
    78.             @Override
    79.             public ObservableSource apply(Observable upstream) {
    80.                 return upstream.subscribeOn(Schedulers.io())  //给上面代码分配异步线程
    81.                         .observeOn(AndroidSchedulers.mainThread()) //给下面代码分配主线程
    82.                         .map(new Function() {
    83.                             @Override
    84.                             public UD apply(UD ud) throws Exception {
    85.                                 Log.d(TAG, "apply: 我监听到你了, 居然在执行");
    86.                                 return ud;
    87.                             }
    88.                         });
    89.             }
    90.         };
    91.     }
    92.  
    93.  
    94.   //2.Observable 被观察者/发布者 起点 创建过程 : new ObservableCreate(){source=自定义source}
    95.   //传入自定义source  → creat(ObservableOnSubscribe soure
    96.   //ObservableCreate extends Observable
    97.   Observable.create(
    98.           // 传入自定义source  → creat(ObservableOnSubscribe soure
    99.           new ObservableOnSubscribe() {
    100.               @Override
    101.               public void subscribe(ObservableEmitter emitter) throws Exception {
    102.                   emitter.onNext("A");
    103.               }
    104.           })
    105.              // ObservableCreate.map
    106.             // ObservableMap extends AbstractObservableWithUpstream
    107.             // AbstractObservableWithUpstream extends Observable implements HasUpstreamObservableSource
    108.             // ObservableSource source();
    109.           .map(new Function() { //发送一次
    110.               @Override
    111.               public Bitmap apply(String s) throws Exception {
    112.                   return null;
    113.               }
    114.           })
    115.           .flatMap(new Function>() { //发送很多次
    116.               @Override
    117.               public ObservableSource apply(Bitmap bitmap) throws Exception {
    118.                   ObservableSource bmp=new Observable() {
    119.                       @Override
    120.                       protected void subscribeActual(Observersuper Bitmap> observer) {
    121.  
    122.  
    123.                       }
    124.                   };
    125.                   return bmp;
    126.               }
    127.           })
    128.           .doOnNext(new Consumer() {
    129.               @Override
    130.               public void accept(Bitmap o) throws Exception {
    131.  
    132.  
    133.               }
    134.           })
    135.           .subscribeOn(Schedulers.io()) //给上面调度 异步线程
    136.           .observeOn(AndroidSchedulers.mainThread()) //给下面调度 主线程
    137.           //3.subscribe 订阅过程 ObservableMap.subscribe
    138.           .subscribe(
    139.                   //1.自定义观察者 Observer 终点
    140.                   new Observer() { //interface 
    141.                       @Override // 谁调subscribe 就在那个线程中,不参与调度
    142.                       public void onSubscribe(Disposable d) {
    143.  
    144.  
    145.                       }
    146.  
    147.  
    148.                       @Override
    149.                       public void onNext(Bitmap aBoolean) {
    150.  
    151.  
    152.                       }
    153.  
    154.  
    155.                       @Override
    156.                       public void onError(Throwable e) {
    157.  
    158.  
    159.                       }
    160.  
    161.  
    162.                       @Override
    163.                       public void onComplete() {
    164.  
    165.  
    166.                       }
    167.                   });
    168.  
    169.  
    170.     }
    171.  
    172.

    自定义source ObservableCreate subscribe订阅 ← 自定义观察者

    CreateEmitter.OnNext() subscribeActual(自定义观察者) ← subscribeActual new Observer()

    ↖ ↘ ↓ 创建发射器

    ↖ ↘CreateEmitter(自定义观察者)---------------------------->onNext();

    ↖ ↓ 调用

    [new ObservableOnSubscribe{}] observer.onSubscribe()

    callback ↖ ↓

    ↖source.subscribe(发射器)

    3> Observable 创建过程时序图如下

    Observable(被观察者) ObservableOnSubscribe RxJavaPlugins ObservableCreate

    | | | |

    |1.new ObservableOnSubscribe() | | |

    |------------------------------------->| | |

    |─┐2.Create | | |

    |<┘ | | |

    | 3.onAssembly(new ObservableCreate()) | |

    |------------------------------------------------------->|4.new ObservableCreate()|

    | | |----------------------->|

    | 6.返回ObservableCreate对象 | 5.ObservableCreate |

    |<-------------------------------------------------------|<-----------------------|

    | | | |

    Observable(被观察者) ObservableOnSubscribe RxJavaPlugins ObservableCreate

    标准观察者模式 RxJava(耦合度低)

    ↙Observer1 被观察者(发布) 抽象层(发射器) map

    Observable ← Observer2 ObservableOnSubscribe CreateEmitter map

    List observers ↖Observer3 source.onNext()------->.onNext() ↓

    add observer ... ↘ 自定义观察者(订阅)

    remove observer Observer

    notify observer .onNext()

    4>ObservableMap源码分析

    一个Map源码分析:

    代码区域 流程区域

    Observable.create | | ObservableCreate ObservableMap Observable

    ↓ | |← subscribeActual(obse) ← subscribeActual(obse) ← .subscribe

    ObservableCreate |ObservableOnSubscribe|

    ↓ | 自定义 source | ↓ ↓ ↓

    ObservableMap | |

    ↓ | |--->CreateEmitter--------->MapObserver--------->自定义Observer

    Observable.subscribe | |

    5>RxJava线程切换

    1.     /**
    2.      * Calls the associated hook function.
    3.      * @param defaultScheduler the hook's input value
    4.      * @return the value returned by the hook
    5.      */
    6.     @NonNull
    7.     public static Scheduler onIoScheduler(@NonNull Scheduler defaultScheduler) {
    8.         Functionsuper Scheduler, ? extends Scheduler> f = onIoHandler;
    9.         //默认是null
    10.         if (f == null) {
    11.             return defaultScheduler;
    12.         }
    13.         return apply(f, defaultScheduler);
    14.     }
    15.  
    16.  
    17.  最终 Scheduler IO                                                                  scheduler.io()
    18.  
    19.  
    20.     /**
    21.      * Sets the specific hook function.
    22.      * @param handler the hook function to set, null allowed
    23.      */
    24.     public static void setIoSchedulerHandler(@Nullable Functionsuper Scheduler, ? extends Scheduler> handler) {
    25.         if (lockdown) {
    26.             throw new IllegalStateException("Plugins can't be changed anymore");
    27.         }
    28.         onIoHandler = handler;
    29.     }
    30.  
    31.  
    32.  
    33.  
    34.  
    35.  
    36.  Schedulers有很多细节
    37.  //TODO 1.Schedulers.io() 线程策略机制
    38.  Schedulers----> Scheduler IO---->new IOTask()--->DEFAULT =new IoScheduler()-->线程池 ExecutorService
    39.  
    40.  
    41.     .subscribeOn(
    42.             //RxJavaPlugins.onIoScheduler(IO); HOOK技术  Calls the associated hook function.
    43.             Schedulers.io() //策略机制
    44.     )
    45.     @NonNull
    46.     public static Scheduler io() {
    47.         return RxJavaPlugins.onIoScheduler(IO);
    48.     }
    49.     IO = RxJavaPlugins.initIoScheduler(new IOTask());
    50.  
    51.  
    52.     static final class IOTask implements Callable {//有返回值的任务 返回Scheduler
    53.         @Override
    54.         public Scheduler call() throws Exception {
    55.             return IoHolder.DEFAULT;
    56.         }
    57.     }
    58.     static final Scheduler DEFAULT = new IoScheduler();
    59.  
    60.  
    61.     public IoScheduler() { this(WORKER_THREAD_FACTORY); }
    62.  
    63.  
    64.     public IoScheduler(ThreadFactory threadFactory) {
    65.         this.threadFactory = threadFactory;
    66.         this.pool = new AtomicReference(NONE); //线程池
    67.         start();
    68.     }
    69.  
    70.  
    71.  //TODO 2.subscribeOn() new IoScheduler---线程池 传入
    72.    public final Observable subscribeOn(Scheduler scheduler) {
    73.        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
    74.        return RxJavaPlugins.onAssembly(new ObservableSubscribeOn(this, scheduler));
    75.    }
    76.  
    77.  
    78.    public ObservableSubscribeOn(ObservableSource source, Scheduler scheduler) {
    79.        super(source);
    80.        this.scheduler = scheduler;
    81.    }
    82.    //在终点 订阅时 会调用 subscribeActual()
    83.    public final void subscribe(Observersuper T> observer) {
    84.         subscribeActual(observer);
    85.    }
    86.  
    87.  
    88.    //这个在ObservableSubscribeOn中
    89.     @Override
    90.     public void subscribeActual(final Observersuper T> s) {
    91.         final SubscribeOnObserver parent = new SubscribeOnObserver(s);
    92.         s.onSubscribe(parent);
    93.         parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
    94.     }
    95.  
    96.  
    97.     final class SubscribeTask implements Runnable {
    98.         private final SubscribeOnObserver parent;
    99.  
    100.  
    101.         SubscribeTask(SubscribeOnObserver parent) {
    102.             this.parent = parent;
    103.         }
    104.  
    105.  
    106.         @Override
    107.         public void run() {
    108.             source.subscribe(parent);
    109.         }
    110.     }
    111.  
    112.  
    113.     //ioScheduler(线程池).scheduleDirect(new SubscribeTask(parent))
    114.     public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
    115.         final Worker w = createWorker(); == IoScheduler.createWorker(){return new EventLoopWorker(pool.get());}
    116.         final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);//自己包装了 Runnable
    117.         DisposeTask task = new DisposeTask(decoratedRun, w);//包装一层 Runnable
    118.         w.schedule(task, delay, unit);//↓ EventLoopWorker.schedule
    119.         return task;
    120.     }
    121.  
    122.  
    123.     public Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
    124.         return threadWorker.scheduleActual(action, delayTime, unit, tasks);//↓
    125.     }
    126.  
    127.  
    128.     @NonNull
    129.     public ScheduledRunnable scheduleActual(final Runnable run, long delayTime,
    130.      @NonNull TimeUnit unit, @Nullable DisposableContainer parent) {
    131.         Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
    132.         ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);
    133.         Future f;
    134.         try {
    135.             if (delayTime <= 0) {
    136.                 f = executor.submit((Callable)sr);//将任务最终交给线程池执行
    137.             } else {
    138.                 f = executor.schedule((Callable)sr, delayTime, unit);
    139.             }
    140.             sr.setFuture(f);
    141.         }
    142.         return sr;
    143.     }
    144.  
    145.  
      146.  
 
       
 
      6>SubscribeOn(Schedulers.io()) 时序图
       
 
       
 
      属于异步线程
       
 
      自定义 ObservableCreate SubscribeTask ObservableSubscribeOn<--subscribe(订阅)<-- 终点
       
 
      source subscribeActual(包裹1) subscribeActual(终点) subscribeActual Observer
       
 
      ↖ ↓ run{ ↓
       
 
      包裹2.onNext source.subXX(包裹2) <--source.subXX() SubscribeOnObserver(Observer)
       
 
      ↘ ↓ } 终点存放
       
 
      CreateEmitter ↓
       
 
      包裹1存放 包裹2 线程池
       
 
      onNext(){ executor.submit((Callable)sr)
       
 
      包裹1.onNext() 这时从线程池中执行异步任务
       
 
      }
       
 
      异步线程
       
 
       
 
      Scheduler MAIN_THREAD <---------------------------- ObserverOn(AndroidSchedulers.mainThread())
       
 
      ↓
       
 
      DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()));-> HandlerScheduler$.HandlerWork
       
 
      scheduler{handler.sendMessage(run)}
       
 
       
 
       
 
       
 
       
 
       

                
                    
                

                
    146. 

                    相关阅读:
      
                    
2022牛客暑期多校训练营7（BCFGJ）                            
      
基于Kinect 动捕XR直播解决方案 - 硬件篇                            
      
mysql innodb 存储引擎                            
      
ubuntu20.04配置解压版mysql5.7                            
      
企业如何做好供应链管理工作？8个步骤及应用详解！                            
      
笔试训练2                            
      
oak深度相机入门教程-人物跟踪                            
      
java项目接口重复提交解决方案                            
      
linux中 struct page 与物理地址的关系                            
      
led护眼灯真的能护眼吗？Led护眼灯的好处                            
      
                          
                
      147. 
                
    147. 
                    原文地址：https://blog.csdn.net/x910131/article/details/126071437
                
      148.