源码分析1
作者:互联网
1、基本的订阅、事件发射、取消订阅
public static void case2() { Observable.create(new ObservableOnSubscribe<Integer>() { @Override public void subscribe(ObservableEmitter<Integer> emitter) throws Exception { Log.d(TAG, "emit 1"); emitter.onNext(1); Log.d(TAG, "emit 2"); emitter.onNext(2); Log.d(TAG, "emit 3"); emitter.onNext(3); Log.d(TAG, "emit onComplete"); emitter.onComplete(); Log.d(TAG, "emit 4"); emitter.onNext(4); } }).subscribe(new Observer<Integer>() { private Disposable mDisposable; private int mCount = 0; @Override public void onSubscribe(Disposable d) { Log.d(TAG, "onSubscribe"); mDisposable = d; } @Override public void onNext(Integer value) { Log.d(TAG, "onNext: value = " + value); mCount++; if (mCount == 2) { Log.d(TAG, "dispose"); mDisposable.dispose(); Log.d(TAG, "isDisposed : " + mDisposable.isDisposed()); } } @Override public void one rror(Throwable e) { Log.d(TAG, "onError: " + e.toString()); } @Override public void onComplete() { Log.d(TAG, "onComplete"); } }); }
2、线程调度
public static void case4() { Observable.create(new ObservableOnSubscribe<Integer>() { @Override public void subscribe(ObservableEmitter<Integer> emitter) throws Exception { Log.d(TAG, "subscribe: thread = " + Thread.currentThread()); emitter.onNext(1); emitter.onComplete(); } }) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Consumer<Integer>() { @Override public void accept(Integer integer) throws Exception { Log.d(TAG, "accept: thread = " + Thread.currentThread()); } }); }
标签:分析,Log,void,public,emitter,TAG,Override,源码 来源: https://www.cnblogs.com/andy-loong/p/11307101.html