2020年7月7日星期二

RxJava2.0（转载）

零、来源

来源：Carson_Ho-简书

一、基础知识

角色作用类比
被观察者（Observable）产生事件顾客
观察者（Observer）接收事件，并给出响应动作厨房
订阅（Subscribe）连接 被观察者 & 观察者服务员
事件（Event）被观察者 & 观察者 沟通的载体菜式


二、基础使用

1.导入连接

implementation 'io.reactivex.rxjava2:rxjava:2.2.19'implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'

2.创建被观察者

 //创建被观察者，产生事件 public Observable<Integer> createObservable() {  Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {   @Override   public void subscribe(ObservableEmitter<Integer> e) throws Exception {    e.onNext(1);    e.onNext(2);    e.onNext(3);    e.onComplete();   }  });  return observable; }

3.创建观察者

	//创建观察者 public Observer<Integer> createObserver() {  Observer<Integer> observer = new Observer<Integer>() {   @Override   public void onSubscribe(Disposable d) {    Log.v("lanjiabinRx", "onSubscribe 连接");   }   @Override   public void onNext(Integer value) {    Log.v("lanjiabinRx", "onNext " + value + " 事件");   }   @Override   public void onError(Throwable e) {    Log.v("lanjiabinRx", "onError 事件");   }   @Override   public void onComplete() {    Log.v("lanjiabinRx", "onComplete 事件");   }  };  return observer; }

4.建立subscribe()连接

 	//观察者订阅被观察者 public void createSubscribe() {  createObservable().subscribe(createObserver()); }

5.调用和结果

 createSubscribe();

image.png

6.链式调用

 	//链式调用 public void chainCall() {  Observable.create(new ObservableOnSubscribe<Integer>() {   @Override   public void subscribe(ObservableEmitter<Integer> e) throws Exception {    e.onNext(1);    e.onNext(2);    e.onNext(3);    e.onComplete();   }  }).subscribe(new Observer<Integer>() {   @Override   public void onSubscribe(Disposable d) {    Log.v("lanjiabinRx", "onSubscribe 连接");   }   @Override   public void onNext(Integer value) {    Log.v("lanjiabinRx", "onNext " + value + " 事件");   }   @Override   public void onError(Throwable e) {    Log.v("lanjiabinRx", "onError 事件");   }   @Override   public void onComplete() {    Log.v("lanjiabinRx", "onComplete 事件");   }  }); }

6.切断连接

即观察者 无法继续 接收 被观察者的事件，但被观察者还是可以继续发送事件

Disposable mDisposable; //1.定义@Overridepublic void onSubscribe(Disposable d) {  mDisposable=d; //2.赋值  Log.v("lanjiabinRx", "onSubscribe 连接");}@Overridepublic void onNext(Integer value) {  if (value==2) mDisposable.dispose(); //3.在第二个next事件断开连接  Log.v("lanjiabinRx", "onNext " + value + " 事件");}

三、创建操作符

0.总图


1. create （基础发送）

最基础的创建

 //1.create public void chainCall() {  Observable.create(new ObservableOnSubscribe<Integer>() {   @Override   public void subscribe(ObservableEmitter<Integer> e) throws Exception {    e.onNext(1);    e.onNext(2);    e.onNext(3);    e.onComplete();   }  }).subscribe(new Observer<Integer>() {   @Override   public void onSubscribe(Disposable d) {    Log.v("lanjiabinRx", "onSubscribe 连接");   }   @Override   public void onNext(Integer value) {    Log.v("lanjiabinRx", "onNext " + value + " 事件");   }   @Override   public void onError(Throwable e) {    Log.v("lanjiabinRx", "onError 事件");   }   @Override   public void onComplete() {    Log.v("lanjiabinRx", "onComplete 事件");   }  }); }

2. just （立刻发送10以下）

  • 快速创建1个被观察者对象（Observable）
  • 发送事件的特点：直接发送传入的事件
  • 最多只能发送十个参数
  • 应用场景：快速创建 被观察者对象（Observable） & 发送10个以下事件
 //2.just public void justDo() {  Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10).subscribe(new Observer<Integer>() {   @Override   public void onSubscribe(Disposable d) {    Log.d("lanjiabinRx", "开始采用subscribe连接");   }   @Override   public void onNext(Integer integer) {    Log.d("lanjiabinRx", "接受的事件 onNext =" + integer);   }   @Override   public void onError(Throwable e) {    Log.d("lanjiabinRx", "接受的事件 onError =" + e.getMessage());   }   @Override   public void onComplete() {    Log.d("lanjiabinRx", "接受的事件 onComplete");   }  }); }

结果：
image.png

3. fromArray （数组发送）

  • 快速创建1个被观察者对象（Observable）
  • 发送事件的特点：直接发送 传入的数组数据
  • 会将数组中的数据转换为Observable对象

应用场景：
1.快速创建 被观察者对象（Observable） & 发送10个以上事件（数组形式）
2.数组元素遍历

 //3.fromArray public void fromArrayDo() {  Integer[] items = {0, 1, 2, 3, 4};  Observable.fromArray(items).subscribe(new Observer<Integer>() {   @Override   public void onSubscribe(Disposable d) {    Log.v("lanjiabinRx", "开始采用subscribe连接");   }   @Override   public void onNext(Integer integer) {    Log.v("lanjiabinRx", "接受的事件 onNext =" + integer);   }   @Override   public void onError(Throwable e) {    Log.v("lanjiabinRx", "接受的事件 onError =" + e.getMessage());   }   @Override   public void onComplete() {    Log.v("lanjiabinRx", "接受的事件 onComplete");   }  }); }

结果：

image.png

4. fromIterable （集合发送）

  • 快速创建1个被观察者对象（Observable）
  • 发送事件的特点：直接发送 传入的集合List数据
  • 会将数组中的数据转换为Observable对象

应用场景：
1.快速创建 被观察者对象（Observable） & 发送10个以上事件（集合形式）
2.集合元素遍历

 //4.fromIterable public void fromIterableDo(){  List<Integer> list = new ArrayList<>();  list.add(1);  list.add(2);  list.add(3);  Observable.fromIterable(list).subscribe(new Observer<Integer>() {   @Override   public void onSubscribe(Disposable d) {    Log.v("lanjiabinRx", "开始采用subscribe连接");   }   @Override   public void onNext(Integer integer) {    Log.v("lanjiabinRx", "接受的事件 onNext =" + integer);   }   @Override   public void onError(Throwable e) {    Log.v("lanjiabinRx", "接受的事件 onError =" + e.getMessage());   }   @Override   public void onComplete() {    Log.v("lanjiabinRx", "接受的事件 onComplete");   }  }); }

结果：

image.png

// 下列方法一般用于测试使用<-- empty() -->// 该方法创建的被观察者对象发送事件的特点：仅发送Complete事件，直接通知完成Observable observable1=Observable.empty(); // 即观察者接收后会直接调用onCompleted（）<-- error() -->// 该方法创建的被观察者对象发送事件的特点：仅发送Error事件，直接通知异常// 可自定义异常Observable observable2=Observable.error(new RuntimeException())// 即观察者接收后会直接调用onError（）<-- never() -->// 该方法创建的被观察者对象发送事件的特点：不发送任何事件Observable observable3=Observable.never();// 即观察者接收后什么都不调用

5. defer （获取最新数据）

  • 直到有观察者（Observer ）订阅时，才动态创建被观察者对象（Observable） & 发送事件
  • 通过 Observable工厂方法创建被观察者对象（Observable）
  • 每次订阅后，都会得到一个刚创建的最新的Observable对象，这可以确保Observable对象里的数据是最新的

应用场景：
动态创建被观察者对象（Observable） & 获取最新的Observable对象数据

 //5.defer Integer i = 10; //第一次赋值 public void deferDo() {  Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<? extends Integer>>() {   @Override   public ObservableSource<? extends Integer> call() throws Exception {    return Observable.just(i);   }  });  i = 15; //第二次赋值  observable.subscribe(new Observer<Integer>() {   @Override   public void onSubscribe(Disposable d) {    Log.v("lanjiabinRx", "开始采用subscribe连接");   }   @Override   public void onNext(Integer integer) {    Log.v("lanjiabinRx", "接受的事件 onNext =" + integer);   }   @Override   public void onError(Throwable e) {    Log.v("lanjiabinRx", "接受的事件 onError =" + e.getMessage());   }   @Override   public void onComplete() {    Log.v("lanjiabinRx", "接受的事件 onComplete");   }  }); }

结果：得到最新赋值的数字，说明取到了最新的数据

image.png

6. timer （延迟发送）

  • 快速创建1个被观察者对象（Observable）
  • 发送事件的特点：延迟指定时间后，发送1个数值0（Long类型）
  • 本质 = 延迟指定时间后，调用一次 onNext(0)

应用场景：
延迟指定事件，发送一个0，一般用于检测

 //6.timer public void timerDo() {  // 注：timer操作符默认运行在一个新线程上  // 也可自定义线程调度器（第3个参数）：timer(long,TimeUnit,Scheduler)  //TimeUnit.SECONDS延迟2s后，发送一个0    /**   * timer(long delay, TimeUnit unit)   * delay 数值   * unit 单位   * 下面就是 2数值，单位为秒，所以是2秒   * */  Observable.timer(2, TimeUnit.SECONDS).subscribe(new Observer<Long>() {   @Override   public void onSubscribe(Disposable d) {    Log.v("lanjiabinRx", "开始采用subscribe连接");   }   @Override   public void onNext(Long aLong) {    /*    * 得到的结果为0 一般用于检测    * */    Log.v("lanjiabinRx", "接受的事件 onNext =" + aLong);   }   @Override   public void onError(Throwable e) {    Log.v("lanjiabinRx", "接受的事件 onError =" + e.getMessage());   }   @Override   public void onComplete() {    Log.v("lanjiabinRx", "接受的事件 onComplete");   }  }); }

结果：

image.png

7. interval （周期发送，无限）

  • 快速创建1个被观察者对象（Observable）
  • 发送事件的特点：每隔指定时间 就发送 事件
  • 发送的事件序列 = 从0开始、无限递增1的的整数序列
 //7.interval public void intervalDo() {  /**   * 从0开始递增   *   * @param initialDelay （Long）   *   初始延迟时间（第一次延迟时间）   * @param period （Long）   *   后续数字发射之间的时间间隔（一个周期时间）   * @param unit   *   时间单位   * */  Observable.interval(3, 2, TimeUnit.SECONDS).subscribe(new Observer<Long>() {   @Override   public void onSubscribe(Disposable d) {    Log.v("lanjiabinRx", "开始采用subscribe连接");   }   @Override   public void onNext(Long aLong) {    Log.v("lanjiabinRx", "接受的事件 onNext =" + aLong);   }   @Override   public void onError(Throwable e) {    Log.v("lanjiabinRx", "接受的事件 onError =" + e.getMessage());   }   @Override   public void onComplete() {    Log.v("lanjiabinRx", "接受的事件 onComplete");   }  }); }

结果：

image.png

8. intervalRange （周期发送，有限，指定数据）

  • 作用类似于interval（），但可指定发送的数据的数量
 	//8.intervalRange public void intervalRangeDo() {  /**   *   * @param start 起始值   * @param count 总共要发送的值的数量，如果为零，则运算符将在初始延迟后发出onComplete   * @param initialDelay 发出第一个值（开始）之前的初始延迟   * @param period 后续值之间的时间段   * @param unit 时间单位   * */  Observable.intervalRange(3, 10, 2, 1, TimeUnit.SECONDS)    .subscribe(new Observer<Long>() {   @Override   public void onSubscribe(Disposable d) {    Log.v("lanjiabinRx", "开始采用subscribe连接");   }   @Override   public void onNext(Long aLong) {    Log.v("lanjiabinRx", "接受的事件 onNext =" + aLong);   }   @Override   public void onError(Throwable e) {    Log.v("lanjiabinRx", "接受的事件 onError =" + e.getMessage());   }   @Override   public void onComplete() {    Log.v("lanjiabinRx", "接受的事件 onComplete");   }  }); }

结果：3-12 经历10个数

image.png

9. range (无延迟，Integer类型指定数据)

  • 作用类似于intervalRange（），但区别在于：无延迟发送事件
	//9.range public void rangeDo(){  /**   * @param start   *   序列中第一个Integer的值   * @param count   *   要生成的顺序整数的数量   * */  Observable.range(3,5).subscribe(new Observer<Integer>() {   @Override   public void onSubscribe(Disposable d) {    Log.v("lanjiabinRx", "开始采用subscribe连接");   }   @Override   public void onNext(Integer integer) {    Log.v("lanjiabinRx", "接受的事件 onNext =" + integer);   }   @Override   public void onError(Throwable e) {    Log.v("lanjiabinRx", "接受的事件 onError =" + e.getMessage());   }   @Override   public void onComplete() {    Log.v("lanjiabinRx", "接受的事件 onComplete");   }  }); }

结果：

image.png

10. rangeLong （无延迟，Long类型指定数据）

 //10.rangeLong public void rangeLongDo(){  /**   * @param start   *   Long类型，序列中第一个Integer的值   * @param count   *   Long类型，要生成的顺序整数的数量   * */  Observable.rangeLong(3,8).subscribe(new Observer<Long>() {   @Override   public void onSubscribe(Disposable d) {    Log.v("lanjiabinRx", "开始采用subscribe连接");   }   @Override   public void onNext(Long aLong) {    Log.v("lanjiabinRx", "接受的事件 onNext =" + aLong);   }   @Override   public void onError(Throwable e) {    Log.v("lanjiabinRx", "接受的事件 onError =" + e.getMessage());   }   @Override   public void onComplete() {    Log.v("lanjiabinRx", "接受的事件 onComplete");   }  }); }

结果：

image.png

编程中我们会遇到多少挫折？表放弃，沙漠尽头必是绿洲。

RxJava2.0（转载）亚马逊礼品卡、 米兰网、 jpgoodbuy、 疫情打击之下，外贸人此时到底该做些什么？、 如何使用亚马逊社交媒体促销代码、 听说2019中东电商平台竟是以"自建站"为主，背后原因竟是？、 珠海长隆高端品质旅游团、 珠海长隆高端品质旅游团、 珠海长隆高端品质旅游团、
发帖者 时间：

没有评论:

发表评论

订阅： 博文评论 (Atom)