rxjava很強大，非常強大。我很喜歡。嘿嘿

剛接觸rxjava不久，整理了下程式碼，留個備份。

首先不清楚觀察者的，建議去網上看下先。不然肯定是懵逼的。  

目錄

/**第一個簡單使用*/
protected abstract void testRxJava();
/**簡化*/
protected abstract void testRxJava1();
/**轉換map*/
protected abstract void testRxJava2();
/**轉換flatmap*/
protected abstract void testRxJava3();
/**just集合操作*/
protected abstract void 
 
testRxJava4();
/**toList() take()用法 */
protected abstract void testRxJava5();
/**interval() take()用法*/
protected abstract void testRxJava6();
/**subscribeOn(Schedulers.io())執行緒執行的問題*/
protected abstract void testRxJava7();

根據目錄，可以匹配對應的功能例子

1.

@Override
protected  void testRxJava() {
    Observable<String> myObservable = Observable.create
 
(new Observable.OnSubscribe<String>(){

        @Override
public void call(Subscriber<? super String> subscriber) {
            subscriber.onNext("被觀察者  傳送的資料");
subscriber.onCompleted();
}
    });
Subscriber<String> mySubscriber = new Subscriber<String>() {
        @Override
 
public void onCompleted() {

        }

        @Override
public void onError(Throwable e) {

        }

        @Override
public void onNext(String s) {
            Log.d("接受到的被觀察者的訊息",s + "testRxJava");
}
    };
/**訂閱 建立聯絡*/
myObservable.subscribe(mySubscriber);
}

2.

@Override
protected void testRxJava1() {
   // Observable<String> myObservable = Observable.just("傳送資料");
    //new Action1 是 new Subscriber的簡寫  只包含call方法   也可以自行了解 Action0 等方法
Observable.just("傳送資料").subscribe(new Action1<String>() {
        @Override
public void call(String s) {
            Log.d("接收到的資料:", s + "testRxJava1");
}
    });
}

3.

@Override
protected void testRxJava2() {
    Observable.just("傳送資料").map(new Func1<String, String>() {

        @Override
public String call(String s) {
            return s + "中途穿插";
}
    }).subscribe(new Action1<String>() {
        @Override
public void call(String s) {
            Log.d("接收到的資料",s + "testRxJava2");
}
    });
}

@Override
protected void testRxJava3() {
    Observable.just("傳送資料").flatMap(new Func1<String, Observable<String>>() {
        @Override
public Observable<String> call(String s) {
            return Observable.just("中間搞下事情");
}
    }).subscribe(new Action1<String>() {
        @Override
public void call(String s) {
            Log.d("接收到的資料",s + "testRxJava3");
}
    });
}

5.

@Override
protected void testRxJava4() {
    List<String> list = new ArrayList<String>();
list.add("a");
list.add("b");
list.add("c");
Observable.just(list).subscribe(new Action1<List<String>>() {
        @Override
public void call(List<String> strings) {
            String result = "";
            for (String item : strings){
                result += item;
}
            Log.d("testRxJava4 結果：",result);
}
    });
/** 注意，just()方法也可以傳list，但是傳送的是整個list物件，而from（）傳送的是list的一個item
     *
     */
Observable.from(list).subscribe(new Action1<String>() {
        @Override
public void call(String s) {
            Log.d("testRxJava4 結果2：",s);
}
    });
Observable.from(list).map(new Func1<String, String>() {
        @Override
public String call(String s) {
            if(s.equals("c")){
                return s + "lahm";
}
            return s;
}
    }).subscribe(new Action1<String>() {
        @Override
public void call(String s) {
            Log.d("testRxJava4 結果3：",s);
}
    });
}

6.

/**
 * toList 總結：可以看到這裡講String[] 陣列型別轉化為List型別了，toList()操作符的目的應該是將所有的資料放進一個List中去，
 * 然後將這個List傳遞給這裡的subscribe。
 *
 * take 只發射前面的N項資料
 */
@Override
protected void testRxJava5() {
    String [] data = new String[]{"a","b","c"};
Observable.from(data).take(2).toList().subscribe(new Action1<List<String>>() {
        @Override
public void call(List<String> strings) {
            String result = "";
            for(String item : strings) result += item;
Log.d("testRxJava5 結果：",result);
}
    });
}

7.

/***
 * *  take 只發射前面的N項資料
 */
@Override
protected void testRxJava6() {
    //這個方法時2秒後輪訓，輪訓的間隔時間是3秒
    //除了  ed限時輸入 驗證碼   還沒發現其他的用法
Observable.interval(2,3, TimeUnit.SECONDS).take(3).subscribeOn(Schedulers.io()).subscribe(new Action1<Long>() {
        @Override
public void call(Long aLong) {
            Log.d("testRxJava6 結果：",aLong+"");
}
    });
}

8.

/***
 * subscribeOn(Schedulers.io()) 執行緒執行的問題
 */
@Override
protected void testRxJava7() {
    Observable.just("測試").map(new Func1<String, String>() {
        @Override
public String call(String s) {
            Log.d("testRxJava7：","map執行的執行緒："+Thread.currentThread().getName());
            return s+"轉換1";
}
    })
            //result   RxIoScheduler-3
    //.subscribeOn(Schedulers.io())
    //main
.subscribeOn(AndroidSchedulers.mainThread())
    .subscribe(new Action1<String>() {
        @Override
public void call(String s) {
            Log.d("testRxJava7：","接收執行的執行緒："+Thread.currentThread().getName());
}
    });
}