欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  移动技术

RxJava2配置及使用详解

程序员文章站 2022-06-04 21:41:45
rxjava2.0是一个非常棒的流式编程,采用的观察者模式思想,事件的产生者产生事间之后发送给绑定的接受者,接受顺序与发送顺序一致. 依赖: compile '...

rxjava2.0是一个非常棒的流式编程,采用的观察者模式思想,事件的产生者产生事间之后发送给绑定的接受者,接受顺序与发送顺序一致.

依赖:

compile 'io.reactivex.rxjava2:rxjava:2.0.1'
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'

简单使用:

//观察者模式,这里产生事件,事件产生后发送给接受者,但是一定要记得将事件的产生者和接收者捆绑在一起,否则会出现错误
observable.create(new observableonsubscribe<string>() {
  @override
  public void subscribe(observableemitter<string> e) throws exception {
    //这里调用的方法会在产生事件之后会发送给接收者,接收者对应方法会收到
    e.onnext("hahaha");
    e.onerror(new exception("wulala"));
    e.oncomplete();
  }/*--
}).subscribe(new observer<string>() {
  //接受者,根据事件产生者产生的事件调用不同方法
  @override
  public void onsubscribe(disposable d) {
    log.e(tag, "onsubscribe: ");
  }

  @override
  public void onnext(string value) {
    log.e(tag, "onnext: " + value);
  }

  @override
  public void onerror(throwable e) {
    log.e(tag, "onerror: ", e);
  }

  @override
  public void oncomplete() {
    log.e(tag, "oncomplete: ");
  }
});

我们来用图解一下这其中发生了什么事:

RxJava2配置及使用详解

上游朝下游发送数据,经过subscribe使上下游产生关系,即达成订阅.

解析1:

observableemitter,这是个啥东西?emitter:顾名思义,即rxjava的发射器,通过这个发射器,即可发送事件-----通过调用onnext,onerror,oncomplete方法发送不同事件.

注意:

虽然rxjava可以进行事件发送,但这并不意味着你可以随便发送,这其中需要遵循一些规则.

onnext:你可以发送无数个onnext,发送的每个onnext接受者都会接收到.

onerror:当发送了onerror事件之后,发送者onerror之后的事件依旧会继续发送,但是接收者当接收到onerror之后就会停止接收事件了.

oncomplete:当发送了oncomplete事件之后,发送者的oncomplete之后的事件依旧会继续发送,但是接收者接收到oncomplete之后就停止接收事件了.

onerror事件和oncomplete事件是互斥的,但是这并不代表你配置了多个onerror和oncomplete一定会崩溃,多个oncomplete是可以正常运行的,但是只会接收到第一个,之后的就不会再接收到了,多个onerror时,只会接收到第一个,第二个会直接造成程序崩溃. 

解析2:

disposable又是个啥东西,翻译之后百度告诉我这东西叫做一次性的,是用来控制发送者和接受者之间的纽带的,默认为false,表示发送者和接受者直接的通信阀门关闭,可以正常通信,在调用dispose()方法之后,阀门开启,会阻断发送者和接收者之间的通信,从而断开连接.

重载方法:

     subscribe();     //表示发送者随意发送数据,接受者什么都不管,什么都不接收.
     subscribe(consumer<? super t> onnext) {}   //只响应onnext()事件,其他的事件忽略.
     subscribe(consumer<? super t> onnext, consumer<? super throwable> onerror) {}     //含义同上
     subscribe(consumer<? super t> onnext, consumer<? super throwable> onerror, action oncomplete) {}     //含义同上
     subscribe(consumer<? super t> onnext, consumer<? super throwable> onerror, action oncomplete, consumer<? super disposable> onsubscribe) {}   //含义同上

解析3:

默认情况下,发送者和接收者都运行在主线程,但是这显然是不符合实际需求的,我们在日常使用中,通常用的最多的就是在子线程进行各种耗时操作,然后发送到主线程进行,难道我们就没有办法继续用这个优秀的库了?想多了你,一个优秀的库如果连这都想不到,怎么能被称为优秀呢,rxjava中有线程调度器,通过线程调度器,我们可以很简单的实现这种效果,下面放代码.

observable.create(new observableonsubscribe<string>() {
  @override
  public void subscribe(observableemitter<string> e) throws exception {
    e.onnext("hahaha");
    e.onnext("hahaha");
    e.onnext("hahaha");
    log.e(tag,"运行在什么线程" + thread.currentthread().getname());
    e.oncomplete();
  }
}).subscribeon(schedulers.newthread())        //线程调度器,将发送者运行在子线程
 .observeon(androidschedulers.mainthread())     //接受者运行在主线程
 .subscribe(new observer<string>() {
  @override
  public void onsubscribe(disposable d) {
    log.e(tag, "onsubscribe: ");
    log.e(tag,"接收在什么线程" + thread.currentthread().getname());
  }

  @override
  public void onnext(string value) {
    log.e(tag, "onnext: " + value);
  }

  @override
  public void onerror(throwable e) {
    log.e(tag, "onerror: ", e);
  }

  @override
  public void oncomplete() {
    log.e(tag, "oncomplete: ");
  }
});

最终结果:

RxJava2配置及使用详解

可以看到我们只加了两行代码,就实现了效果,还有比这个更优秀,更简单的么?

注意事项:

subscribeon(),只有在第一次调用的时候生效,之后不管调用多少次,只会以第一次为准.

observeon(),可以被调用多次,每次调用都会更改线程.

rxjava线程池中的几个线程选项

  1.  - schedulers.io()      io操作的线程, 通常io操作,如文件读写.
  2.  - schedulers.computation()      计算线程,适合高计算,数据量高的操作.
  3.  - schedulers.newthread()      创建一个新线程,适合子线程操作.
  4. - androidschedulers.mainthread()      android的主线程,主线程

本文介绍简单使用,以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。