java rx_RxJava 和 RxAndroid 一 (基础)
1、RxJava 项目地址
2、RxAndroid 项目地址
3、RxJava 和 RxAndroid 的关系
RxAndroid是RxJava的一个针对Android平台的扩展，主要用于 Android 开发
4、RxJava和EventBus的区别？
5、RxAndroid的使用方法
compile 'io.reactivex:rxandroid:1.2.0'
6、如何查看RxAndroid最新版本？
7、RxAndroid具体使用方法
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 observer = new Observer() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(String s) {
}
} ;
// Subscriber 继承 Observer ，对Observer类做了扩展
Subscriber subscriber = new Subscriber() {
@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(): 这是 Subscriber 增加的方法。它会在 subscribe 刚开始，而事件还未发送之前被调用，可以用于做一些准备工作，例如数据的清零或重置。这是一个可选方法，默认情况下它的实现为空。需要注意的是，如果对准备工作的线程有要求(例如弹出一个显示进度的对话框，这必须在主线程执行)， onStart() 就不适用了，因为它总是在 subscribe 所发生的线程被调用，而不能指定线程。要在指定的线程来做准备工作，可以使用 doOnSubscribe() 方法，具体可以在后面的文中看到。
// Subscriber 继承 Observer ，对Observer类做了扩展
Subscriber subscriber = new Subscriber() {
@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 observable = Observable.create(new Observable.OnSubscribe() {
@Override
public void call(Subscriber super String> subscriber) {
subscriber.onNext( "aa" ) ;
subscriber.onNext( "bb" ) ;
subscriber.onNext( "cc" ) ;
subscriber.onCompleted();
}
});
//just方式 最多支持10个数据
Observable observable1 = Observable.just( "aa" , "bb" , "cc") ;
// 将会依次调用：
// onNext("aa");
// onNext("bb");
// onNext("cc");
// onCompleted();
//from方式
//1:集合
List list = new ArrayList<>() ;
list.add( "aa" ) ;
list.add( "bb" ) ;
list.add( "cc" ) ;
Observable observable2 = Observable.from( list ) ;
//2：数组
String[] words = { "aa", "bb", "cc" };
Observable 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的封装版
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> subjectMapper = new ConcurrentHashMap<>();
@SuppressWarnings("unchecked")
public Observable register(@NonNull Object tag, @NonNull Class clazz) {
List subjectList = subjectMapper.get(tag);
if (null == subjectList) {
subjectList = new ArrayList<>();
subjectMapper.put(tag, subjectList);
}
Subject 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 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 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 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() {
@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 进行线程切换