1.RxJava2的基本介紹

1.1核心思想：

• 觀察者模式
  觀察者(接收器)
  被觀察者(發(fā)射器)
• 響應(yīng)式編程
  基于異步數(shù)據(jù)流的編程模式，數(shù)據(jù)流就像一條河：它可以被觀測，被過濾，被操作，或者為新的消費(fèi)者與另外一條流合并為一條新的流

1.2 簡單使用

• create
  產(chǎn)生一個(gè)觀察者對(duì)象

Observable.create(new ObservableOnSubscribe<Object>() {
            @Override
            public void subscribe(ObservableEmitter<Object> emitter) throws Exception {
                  emitter.onNext("發(fā)送事件");
            }
        });

1.3 操作符

在被觀察者處理數(shù)據(jù)后會(huì)調(diào)用觀察者的相關(guān)方法(即被觀察者發(fā)送事件給觀察者)。而在這個(gè)處理過程中，可以使用rxjava內(nèi)置的方法來進(jìn)行數(shù)據(jù)的過濾處理

• compose
  需要配合transformer使用，將obervableA轉(zhuǎn)化為obervableB，即保證鏈?zhǔn)秸{(diào)用的完整性而不需要中斷。

getRxApi().login(account, password)
                .compose(RxSchedulers.io_main())//傳入Transformer對(duì)象
                .subscribe(observable);

public static ObservableTransformer io_main() {
       return new ObservableTransformer() {
           @Override
           public ObservableSource apply(Observable upstream) {
               return upstream
                       .subscribeOn(Schedulers.io())
                       .unsubscribeOn(Schedulers.io())
                       .observeOn(AndroidSchedulers.mainThread());
           }
       };
    }

• interval
  在指定的時(shí)間間隔持續(xù)發(fā)送從0開始遞增的事件

Observable.interval(1,2, TimeUnit.SECONDS);// 0在1秒后發(fā)送，之后每隔2秒發(fā)送一次事件(1,2....)

• take
  發(fā)送指定

 Observable.interval(1，2, TimeUnit.SECONDS)
        .take(3);//發(fā)送3次事件就終止(1,2,3)

• map

   Observable.interval(0, TimeUnit.SECONDS)
        .map(new Function<Long, String>() {
            @Override
            public String apply(Long aLong) throws Exception {
                return String.valueOf(aLong);//將Long型元素轉(zhuǎn)化為字符串
            }
        });

• flatMap

//有多個(gè)學(xué)生，每個(gè)學(xué)生選修了多門課程。打印出每個(gè)學(xué)生選修的課程名稱
Observable.from(students)
                .flatMap(new Func1<Student, Observable<Course>>() {
                    @Override
                    public Observable<Course> call(Student student) {
                        return Observable.from(student.getCoursesList());
                    }
                })
                .subscribe(new Action1<Course>() {
                    @Override
                    public void call(Course course) {
                        Log.i(TAG, course.getName());
                    }
                });

• just
  與create不同的是，它會(huì)在創(chuàng)建被觀察者對(duì)象之后直接調(diào)用onNext方法

Observable.just("發(fā)送數(shù)據(jù)");

• combineLatest
  將兩個(gè)發(fā)射器的結(jié)果進(jìn)行合并(多用于表單校驗(yàn))

    //結(jié)合Rxbinding使用
    Observable<CharSequence> ObservableName  = RxTextView.textChanges(etUsername);
    Observable<CharSequence> ObservablePassword = RxTextView.textChanges(etPassword);
    Observable.combineLatest(ObservableName, ObservablePassword, new BiFunction<CharSequence, CharSequence, Boolean>() {
            @Override
            public Boolean apply(CharSequence charSequence, CharSequence charSequence2) throws Exception {
                String passowrd = inputPassword.getEditText().getText().toString().trim();
                return !TextUtils.isEmpty(inputUsername.getEditText().getText().toString().trim())
                        && !TextUtils.isEmpty(passowrd);
            }
        }).subscribe(new Consumer<Boolean>() {
            @Override
            public void accept(Boolean aBoolean) throws Exception {
                tvLogin.setEnabled(aBoolean);
            }
        });