1、RxJava 項目位址
https://github.com/ReactiveX/RxJava
2、RxAndroid 項目位址
https://github.com/ReactiveX/RxAndroid
3、RxJava 和 RxAndroid 的關系
RxAndroid是RxJava的一個針對Android平台的擴充,主要用于 Android 開發
4、RxJava和EventBus的差別?
https://www.zhihu.com/question/32179258/answer/54989242
5、RxAndroid的使用方法
compile 'io.reactivex:rxandroid:1.2.0'
6、如何檢視RxAndroid最新版本?
http://search.maven.org/#search%7Cga%7C1%7Ca%3A%22rxandroid%22
7、RxAndroid具體使用方法
http://gank.io/post/560e15be2dca930e00da1083#toc_14
http://blog.csdn.net/theone10211024/article/details/50435325
http://huxian99.github.io/tags/RxJava/
https://github.com/mcxiaoke/RxDocs
8、建立觀察者
package lib.com.myapplication;
import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import rx.Observer;
import rx.Subscriber;
public class Main2Activity extends AppCompatActivity {
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main2);
//建立觀察者 2 種方法
Observer<String> observer = new Observer<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
}
} ;
// Subscriber 繼承 Observer ,對Observer類做了擴充
Subscriber<String> subscriber = new Subscriber<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
}
} ;
}
}
- 從上文可以看到,Subscriber繼承Observer, 隻是 Subscriber對Observer做了一些擴充。Subscriber的使用和Observer完全一樣。
- Subscriber 多了一個 onStart 方法
-
: 這是onStart()
增加的方法。它會在 subscribe 剛開始,而事件還未發送之前被調用,可以用于做一些準備工作,例如資料的清零或重置。這是一個可選方法,預設情況下它的實作為空。需要注意的是,如果對準備工作的線程有要求(例如彈出一個顯示進度的對話框,這必須在主線程執行),Subscriber
就不适用了,因為它總是在 subscribe 所發生的線程被調用,而不能指定線程。要在指定的線程來做準備工作,可以使用onStart()
方法,具體可以在後面的文中看到。doOnSubscribe()
// Subscriber 繼承 Observer ,對Observer類做了擴充
Subscriber<String> subscriber = new Subscriber<String>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
}
@Override
public void onStart() {
super.onStart();
}
} ;
9、建立被觀察者
//create方式
Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext( "aa" ) ;
subscriber.onNext( "bb" ) ;
subscriber.onNext( "cc" ) ;
subscriber.onCompleted();
}
});
//just方式 最多支援10個資料
Observable<String> observable1 = Observable.just( "aa" , "bb" , "cc") ;
// 将會依次調用:
// onNext("aa");
// onNext("bb");
// onNext("cc");
// onCompleted();
//from方式
//1:集合
List<String> list = new ArrayList<>() ;
list.add( "aa" ) ;
list.add( "bb" ) ;
list.add( "cc" ) ;
Observable<String> observable2 = Observable.from( list ) ;
//2:數組
String[] words = { "aa", "bb", "cc" };
Observable<String> observable3 = Observable.from( words ) ;
- Call()方法:當
被訂閱的時候,Observable
的OnSubscribe
方法會自動被調用,事件序列就會依照設定依次觸發(對于上面的代碼,就是觀察者call()
将會被調用三次Subscriber
和一次onNext()
)。這樣,由被觀察者調用了觀察者的回調方法,就實作了由被觀察者向觀察者的事件傳遞,即觀察者模式。onCompleted()
10、訂閱
由于觀察者可以由兩種方式被建立,是以訂閱的方式也有兩種
observable.subscribe( observer ) ;
observable.subscribe( subscriber ) ;
-
的内部實作是這樣的(僅核心代碼):Observable.subscribe(Subscriber)
// 注意:這不是 subscribe() 的源碼,而是将源碼中與性能、相容性、擴充性有關的代碼剔除後的核心代碼。 // 如果需要看源碼,可以去 RxJava 的 GitHub 倉庫下載下傳。 public Subscription subscribe(Subscriber subscriber) { subscriber.onStart(); onSubscribe.call(subscriber); return subscriber; }
- 在subscribe() 中,首先會調用 onStart() 方法,這個方法前文已經介紹了,是可選的。接着會調用 call()方法,我們已經分析了在call()方法中會調用多次 onNext() ,最後調用 onCompleted().看到這裡你就會突然明白原來subscribe() 方法其實相當于依次執行了:onStart() --> onNext()--> onCompleted()
- 從這也可以看出,在 RxJava 中,
并不是在建立的時候就立即開始發送事件,而是在它被訂閱的時候,即當Observable
方法執行的時候。subscribe()
-
和Observer
具有相同的角色,而且Subscriber
在Observer
過程中最終會被轉換成subscribe()
對象Subscriber
- 将傳入的
作為Subscriber
傳回。這是為了友善Subscription
.unsubscribe()
11、RxBus
你是否聽說過EventBus , 他是android 中的事件總線。用rxjava同樣可以實作android的事件總線功能,也就是RxBus.
關于rxbus 的基本說明在這裡 http://nerds.weddingpartyapp.com/tech/2014/12/24/implementing-an-event-bus-with-rxjava-rxbus/
然而這并沒有什麼卵用 !
下面是RxBus的封裝版
package lib.com.myapplication;
import android.support.annotation.NonNull;
import android.util.Log;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import rx.Observable;
import rx.subjects.PublishSubject;
import rx.subjects.Subject;
/**
* Created by ${zyj} on 2016/5/6.
*/
public class RxBus {
private static final String TAG = RxBus.class.getSimpleName();
private static RxBus instance;
public static boolean DEBUG = false;
public static RxBus get() {
if (instance == null) {
synchronized (RxBus.class) {
if (instance == null) {
instance = new RxBus();
}
}
}
return instance;
}
private RxBus() {
}
private ConcurrentHashMap<Object, List<Subject>> subjectMapper = new ConcurrentHashMap<>();
@SuppressWarnings("unchecked")
public <T> Observable<T> register(@NonNull Object tag, @NonNull Class<T> clazz) {
List<Subject> subjectList = subjectMapper.get(tag);
if (null == subjectList) {
subjectList = new ArrayList<>();
subjectMapper.put(tag, subjectList);
}
Subject<T, T> subject;
subjectList.add(subject = PublishSubject.create());
if (DEBUG) Log.d(TAG, "[register]subjectMapper: " + subjectMapper);
return subject;
}
public void unregister(@NonNull Object tag, @NonNull Observable observable) {
List<Subject> subjects = subjectMapper.get(tag);
if (null != subjects) {
if ( observable != null && subjects.contains( observable )){
subjects.remove((Subject) observable);
}
if (isEmpty(subjects)) {
subjectMapper.remove(tag);
}
}
if (DEBUG) Log.d(TAG, "[unregister]subjectMapper: " + subjectMapper);
}
public void post(@NonNull Object content) {
post( content.getClass().getName(), content);
}
@SuppressWarnings("unchecked")
public void post(@NonNull Object tag, @NonNull Object content) {
List<Subject> subjectList = subjectMapper.get(tag);
if (!isEmpty(subjectList)) {
for (Subject subject : subjectList) {
subject.onNext(content);
}
}
if (DEBUG) Log.d(TAG, "[send]subjectMapper: " + subjectMapper);
}
private boolean isEmpty(Collection collection) {
return null == collection || collection.isEmpty();
}
}
RxBus的使用
package lib.com.myapplication;
import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import rx.Observable;
import rx.functions.Action1;
public class Activity1 extends AppCompatActivity {
String tag = "tag" ;
Observable<String> ob ;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity1);
//建立被觀察者
ob = RxBus.get().register( tag , String.class ) ;
//訂閱觀察事件
ob.subscribe(new Action1<String>() {
@Override
public void call(String s) {
System.out.println( "fff-- " + s );
}
}) ;
//發送内容
RxBus.get().post( tag , "我是内容" );
}
@Override
protected void onDestroy() {
super.onDestroy();
//取消訂閱
RxBus.get().unregister( tag , ob );
}
}
- 在Activity銷毀的時候,要取消訂閱服務 。 否則 post() 次數會随着post()調用逐漸增加
- 除了上面的簡單使用外,還可以使用 Schedulers 、AndroidSchedulers 進行線程切換