關於RxJava在業務上的一些思考
最近在工作中，頻繁的使用了Rxjava來解決一些問題，在使用過程中也給予了自己一些思考，如何使用好RxJava，在什麼樣的場景中才能發揮它更好的作用，如何脫離程式碼來理解RxJava的工作機制，下面是自己一些淺顯的思考。
示例
太多示例喜歡鏈式的把RxJava的流程表述起來，這個地方我把觀察者和訂閱者拆開來看。
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("123");
}
});
Observer observer = new Observer<String>() {
...
@Override
public void onNext(String s) {
Log.i("TAG", "onNext: " + s);
}
};
observable.subscribe(observer);
複製程式碼
這個簡單的例子大家應該都知道，只要subcribe產生了訂閱，onNext方法將會收到 emitter.onNext("123"); 發射出去的資料。
這個地方讓我產生思考主要是有一次去吃自助餐，大家在打酸奶的時候，都會拿著一個杯子對準出口，然後按住開關，酸奶就會自動流到杯子中。在這個過程中，我們不妨把酸奶機看做 Observable ,酸奶機裡面的酸奶是許許多多的 emitter.onNext("123") ,按住開關的那一刻產生了 subscribe 訂閱，然後我們是用杯子 Observer 去接牛奶的，當然，我們還有橙子機、酸梅湯機等，則機子內盛的飲料型別就是 Observable<String> 。我們知道，酸奶機有很多個開關入口，這時候，又來一個人，拿著杯子 Observer 來打牛奶，那麼，我和他是一塊共享這酸奶機裡面的酸奶，我們倆都能接收到酸奶，等我們不需要接酸奶了，我們就dispose關閉開關。
eg：
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> observableEmitter) throws Exception {
//模擬耗時任務
for (int i = 0; i < 5; i++) {
observableEmitter.onNext(""+i);
}
}
}).subscribeOn(Schedulers.io());
//杯子1
observable.subscribe(observer1);
//杯子2
observable.subscribe(observer2);
複製程式碼
result:
observer1 onNext: 0 observer2 onNext: 0 observer1 onNext: 1 observer2 onNext: 1 observer1 onNext: 2 observer2 onNext: 2 observer1 onNext: 3 observer2 onNext: 3 observer1 onNext: 4 observer2 onNext: 4 複製程式碼
事件驅動的思考
之前在思考事件驅動這一塊，如何更好的通知其他業務元件，業界比較有名的當屬EventBus，但EventBus用起來很雜亂無章，當專案規模大起來，業務複雜起來時，都不敢修改這個post，雖然解耦了，但事件變得更亂了，所以，自己重新思考了事件驅動這一塊。
鑑於EventBus提供的的思路，我打算用RxJava的方式來實現。以酸奶機為例，當前頁面我想訂閱一個事件，等待被觸發，我完全可以先準備一個杯子(Observer)，然後將他們存到一個集合裡面，待酸奶機(Observable)裡面有酸奶了(observableEmitter.onNext),然後訂閱(subcribe)這個杯子的集合，將酸奶倒到杯子裡，鑑於此思路，用程式碼大致的實現下。
List<Observer> list = new ArrayList<>();
//註冊事件
public void registerObserver(Observer observer) {
list.add(observer);
}
//驅動事件
public void postEvent() {
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("123");
}
});
for (int i = 0; i < list.size(); i++) {
observable.subscribeOn(Schedulers.io()).subscribe(list.get(i));
}
}
@Test
public void Test() {
Observer observer = new Observer<String>() {
...
@Override
public void onNext(String s) {
System.out.println("onNext: " + s);
}
};
//註冊事件
registerObserver(observer);
//傳送事件
postEvent();
}
複製程式碼
這裡只給了大致的思路，用了一個平時都可見的例子來實現了事件驅動。
非同步回撥的思考
最近有一個業務場景，需要監聽RTK當前狀態的變化，業務場景是：
有一個前提，RTK必須先設定賬戶，才能使用後續的服務。如果使用者按照應用準則走，進入主頁後，先去設定頁面設定RTK，然後回到主頁，進入任務執行頁，這時候監聽RTK state是可用的；但如果使用者忘了設定的步驟，或是有強迫症的使用者，我就不往你提示的方式走，我就要先進任務執行員頁，這時候RTK State監聽是不可用的，我們會引導使用者進入設定頁面設定RTK，這個地方又要分情況，如果使用者設定好RTK賬戶，然後返回了任務頁，那麼任務頁的RTK state監聽到了使用者設定了賬戶就會返回可用，那麼這次任務是可使用的；但如果使用者設定好了賬戶，想去診斷RTK當前連線的狀態的話，RTK state監聽事件就會被診斷RTK頁面給設定，也就意味著，任務頁的RTK state監聽就會失效，那麼返回任務頁的話，任務頁是不會有任何反應的。但這一塊也是有解決辦法的，就是任務頁的RTK state監聽放在 onResume 方法裡面，即設定頁面返回任務頁後，觸發任務頁的 onResume 方法，重新奪回RTK state的監聽。 辦法都是有的，但依賴生命週期去做到這點，感覺並不是特別的可靠，我們可以參考上面事件驅動的例子，將RTK state看成是酸奶機，然後哪個頁面(杯子)需要RTK state資訊的話，就可以訂閱(subcribe)酸奶機，如果想要牛奶的話，就post一個資訊出去，告訴酸奶機我要酸奶，下面，我給出一份示例：
Set<Observer> set = new HashSet<>();
//模擬一個RTK state 單例
public Observable getObservableInstance() {
return Observable.create(new ObservableOnSubscribe<RTKState>() {
@Override
public void subscribe(ObservableEmitter<RTKState> emitter) throws Exception {
RTK rtk = DjiSettingUtils.getRTK();
rtk.setStateCallback(new RTKState.Callback() {
@Override
public void onUpdate(@NonNull RTKState rtkState) {
emitter.onNext(rtkState);
}
});
}
});
}
//驅動事件
public void postEvent(Observer observer) {
if (!set.contains(observer)) {
getObservableInstance().subscribeOn(Schedulers.io()).subscribe(observer);
}
}
@Test
public void onCreate() {
Observer observer = new Observer<RTKState>() {
...
@Override
public void onNext(RTKState s) {
System.out.println("onNext: " + s.isRTKBeingUsed());
}
};
//傳送事件
postEvent(observer);
}
複製程式碼
之後，我們只需要關注 onCreate 方法，在任務頁我們發起一個訂閱事件，接收RTK state資訊，在診斷頁面也發起一個訂閱，接收RTK資訊，這樣就不會像上面那樣，搶斷監聽事件的問題。
多圖上傳的思考
業務場景中有需要從無人機中讀取縮圖，並將縮圖上傳至伺服器，圖片上傳我們使用的是七牛雲，因為一次任務產生的縮圖非常多，基本上都在百張左右，我們不可能為了在上傳過程中，因為某些原因導致了斷開了，讓使用者重新上傳所有的縮圖，所以，我們打算讓百張縮圖採用順序上傳，當哪個節點發生錯誤的時候，記住index，等使用者點選重新上傳時，我們再從index的位置繼續上傳，如果按照傳統方式來做的話，第一張上傳成功後，如何通知第二張上傳呢，我這裡給個大致的程式碼：
List<File> list=new ArrayList<>();
int index=0;
public void uploadPic(){
uploadManager.put(list.get(index), key, token, new UpCompletionHandler() {
@Override
public void complete(String key, ResponseInfo info, JSONObject res) {
if (info.isOK()) {
index++;
uploadPic()
} else {
//彈框提示使用者，當前index上傳失敗
}
}
}, null);
}
@Test
public void test(){
uploadPic()
}
複製程式碼
每次上傳成功後都呼叫自身的方法，如果上傳失敗了，則記住index的位置，提示使用者，使用者點選重試上傳，那麼就繼續呼叫 uploadPic 方法，上傳的拿到的檔案還是從index位置開始拿，所以，也是沒有任何問題的。 但是，總覺得這麼設計不那麼優雅，比如我想知道上傳進度的話，那也就意味著我需要在index++方法下面加一個設定進度條的功能，那如果業務需要再加一個上傳完成的操作的話，那是不是又要在index++下面多加一個 index==list.size() 的判斷呢，其實，這樣設計下去的話，整個上傳功能就變得特別的鬆散，移植性也不強，所以，是時候發揮RxJava的 Observer 了。
鑑於非同步回撥的思考，我打算把上傳任務封裝成一個 ObservableOnSubcribe ，每次執行任務成功後，就將事件流onNext交給下游，告訴他我完成了一次上傳，如果上傳失敗了，則發射onError異常。
public class QiNiuBitmapOnSubscribe implements ObservableOnSubscribe<QiniuParam> { ... @Override public void subscribe(final ObservableEmitter<QiniuParam> emitter) throws Exception { //上傳操作 uploadManager.put(file, key, token, new UpCompletionHandler() { @Override public void complete(String key, ResponseInfo info, JSONObject res) { if (info.isOK()) { emitter.onNext(new QiniuParam(key, info, res)); emitter.onComplete(); } else { emitter.onError(new ServerException(-1, res.toString())); } } }, null); } } 複製程式碼
由於圖片是儲存在一個集合中，那麼就肯定要用到RxJava的 fromIterable 來遍歷集合，由於需要保證圖片是有序上傳，就需要用到 concatMap 操作符 , 所以，大致程式碼如下
Observable.fromIterable(fileList)
.concatMap(new Function<QiNiuFile, ObservableSource<QiniuParam>>() {
@Override
public ObservableSource<QiniuParam> apply(QiNiuFile qiniuFile) throws Exception {
//返回七牛雲上傳
return Observable.create(new QiNiuFileOnSubscribe(uploadManager,
qiniuFile.getFile(), qiniuFile.getKey(), qiniuFile.getUploadToken()));
}
}).subscribe(new Observer<QiniuParam>() {
...
@Override
public void onNext(QiniuParam qiniuParam) {
index++;
//通知上傳進度
uploadCallBack.onUploadProcess(index);
}
@Override
public void onError(Throwable e) {
//通知斷傳的位置
uploadCallBack.onUploadQiNiuError(index);
}
@Override
public void onComplete() {
//上傳成功
uploadCallBack.onUploadQiNiuComplete();
}
});
複製程式碼
對於 Observer 來說，他是一個乾淨的接收流，他不關心上游發生的事情，只專注結果的處理。
思考
以上思考有的地方可能不是特別的完善，還需要多思考，RxJava用的人確實很多，但要想玩的溜的話，確實任重而道遠。