Rxjava RxAndroid_io.reactivex.rxjava2:rxandroid:2.1.1

官网：

api 'io.reactivex.rxjava2:rxandroid:2.1.1'
api 'io.reactivex.rxjava2:rxjava:2.2.12'

RxJava：

1、引入java 1.8，或者在根目录做统一管理也可以

2、被观察者跟观察者的处理

   Observable.just("123").subscribe {
            Log.e("msg",it)
        }

 

3、变换（操作符）：把一个事件转换为另一个事件

Map

普通

Observable.just("你是")
  .map(new Func1<String, String>() {
      @Override
      public String call(String s) {
          return s + " 张三疯";
      }
  })
  .subscribe(s -> System.out.println(s));

lambda

  Observable.just("你是:","还是：")
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .map { "${it}张三疯" }
            .subscribe { Log.e("msg",it) }

 

from

 

flatMap

 

filter()：滤掉那些不满足检查条件的。

 

take(): 输出最多指定数量的结果

 

4、调度器（线程切换）

 

 

5、RxAndroid：

AndroidObservable

图片加载：

AndroidObservable.bindActivity(this, retrofitService.getImage(url)) .subscribeOn(Schedulers.io()) 
.subscribe(bitmap -> myImageView.setImageBitmap(bitmap);

广播监听网路变换：

IntentFilter filter = new IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION);
AndroidObservable.fromBroadcast(context, filter)
    .subscribe(intent -> handleConnectivityChange(intent));

View的监听：

ViewObservable.clicks(mCardNameEditText, false)
    .subscribe(view -> handleClick(view));

 

RxView（RxTextView、RxRadioGroup....）：防止按钮重复点击、响应式的界面等

RxView.clicks(button)
              .throttleFirst(1, TimeUnit.SECONDS)
              .subscribe(new Observer<Object>() {
                  @Override
                  public void onCompleted() {
                        log.d ("completed");
                  }
 
                  @Override
                  public void onError(Throwable e) {
                        log.e("error");
                  }
 
                  @Override
                  public void onNext(Object o) {
                       log.d("button clicked");
                  }
              });

 

RxBus：消息传递

 

combineLatest:合并节点-所有输入信息（邮箱、密码、电话号码等）合法才点亮注册按钮

 

merge：合并两个数据源

 

使用concat和first做缓存

 

定时器

Observable.timer(2, TimeUnit.SECONDS)
              .subscribe(new Observer<Long>() {
                  @Override
                  public void onCompleted() {
                      log.d ("completed");
                  }
 
                  @Override
                  public void onError(Throwable e) {
                      log.e("error");
                  }
 
                  @Override
                  public void onNext(Long number) {
                      log.d ("hello world");
                  }
              });

interval:周期性操作

例如：每隔2秒输出日志“helloworld”。

 Observable.interval(2, TimeUnit.SECONDS)
              .subscribe(new Observer<Long>() {
                  @Override
                  public void onCompleted() {
                     log.d ("completed");
                  }
 
                  @Override
                  public void onError(Throwable e) {
                     log.e("error");
                  }
 
                  @Override
                  public void onNext(Long number) {
                     log.d ("hello world");
                  }
              });

schedulePeriodically做轮询请求

Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(final Subscriber<? super String> observer) {
 
                Schedulers.newThread().createWorker()
                      .schedulePeriodically(new Action0() {
                          @Override
                          public void call() {
                              observer.onNext(doNetworkCallAndGetStringResult());
                          }
                      }, INITIAL_DELAY, POLLING_INTERVAL, TimeUnit.MILLISECONDS);
            }
        }).subscribe(new Action1<String>() {
            @Override
            public void call(String s) {
                log.d("polling….”));
            }
        })