RxJava 原理篇

一、框架思想

  • 觀察者模式

    • 觀察者自下而上注入被觀察者
    • 被觀察者自上而下發(fā)射事件


      觀察者模式
  • 裝飾器模式

    • 自上而下,被觀察者被一層層裝飾
    • 自下而上,觀察者被一層層裝飾


      裝飾者模式
  • 策略模式

    • 函數(shù)式編程

二、最簡單的訂閱模型

以下是由just方法和subscribe方法組成的一個(gè)最簡單的訂閱模型

Observable
    .just("Hello RxJava")
    .subscribe(new Observer<String>() {
        @Override
        public void onSubscribe(Disposable d) {
        }
        @Override
        public void onNext(String s) {
            Log.d(TAG, s);
        }
        @Override
        public void onError(Throwable e) {
        }
        @Override
        public void onComplete() {
        Log.d(TAG, "Complete");
        }
    });

對(duì)于just方法

public static <T> Observable<T> just(T item) {
    // 參數(shù)合法判斷欲芹,不允許傳入空item
    ObjectHelper.requireNonNull(item, "The item is null");
    // hook裝飾
    return RxJavaPlugins.onAssembly(
        // 返回真正的ObservableJust對(duì)象
        new ObservableJust<T>(item));
}
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
    Function<? super Observable, ? extends Observable> f = onObservableAssembly;
    if (f != null) {
        // 如果hook方法存在,則調(diào)用hook方法
        return apply(f, source);
    }
    return source;
}
public final class ObservableJust<T> extends Observable<T> implements ScalarCallable<T> {
    private final T value;
    public ObservableJust(final T value) {
        // 保存原始的value
        this.value = value;
    }
    
    // 實(shí)際的訂閱方法的實(shí)現(xiàn)
    @Override
    protected void subscribeActual(Observer<? super T> downStream) {
        // 傳入下游和value
        ScalarDisposable<T> sd = new ScalarDisposable<T>(downStream, value);
        // 訂閱
        downStream.onSubscribe(sd);
        // Runnable.run()吟吝;
        sd.run();
    }
}
public static final class ScalarDisposable<T> extends AtomicInteger implements QueueDisposable<T>, Runnable {
    // 省略很多跟隊(duì)列使用菱父,線程安全相關(guān)的方法
    @Override
    public void run() {
        if (get() == START && compareAndSet(START, ON_NEXT)) {
            downStream.onNext(value);
            if (get() == ON_NEXT) {
                lazySet(ON_COMPLETE);
                downStream.onComplete();
            }
        }
    }
}

對(duì)于subscribe方法

public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
        Action onComplete, Consumer<? super Disposable> onSubscribe) {
    // 參數(shù)合法檢查
    ObjectHelper.requireNonNull(onNext, "onNext is null");
    ObjectHelper.requireNonNull(onError, "onError is null");
    ObjectHelper.requireNonNull(onComplete, "onComplete is null");
    ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");
    // 把4個(gè)Lambda表達(dá)式合并創(chuàng)建LambdaObserver
    LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);
    // 執(zhí)行訂閱
    subscribe(ls);
    // 返回
    return ls;
}
    
@Override
public final void subscribe(Observer<? super T> downStream) {
    // 檢查參數(shù)合法性
    ObjectHelper.requireNonNull(downStream, "observer is null");
    try {
        // hook
        downStream = RxJavaPlugins.onSubscribe(this, downStream);
        // hook后再次檢查Observer不為null
        ObjectHelper.requireNonNull(downStream, "The RxJavaPlugins.onSubscribe hook returned a null Observer. Please change the handler provided to RxJavaPlugins.setOnObservableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");
        // 執(zhí)行實(shí)際的subscribe,傳入下游Observer
        subscribeActual(downStream);
    } catch (NullPointerException e) { // NOPMD
        throw e;
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        // can't call onError because no way to know if a Disposable has been set or not
        // can't call onSubscribe because the call might have set a Subscription already
        RxJavaPlugins.onError(e);
        NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
        npe.initCause(e);
        throw npe;
    }
}

三剑逃、操作符的實(shí)現(xiàn)

最簡單的操作符 map

public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
    // 檢查參數(shù)合法性
    ObjectHelper.requireNonNull(mapper, "mapper is null");
    // hook
    return RxJavaPlugins.onAssembly(
        // 返回真正的 ObservableMap浙宜,傳入了上游引用和mapper方法
        new ObservableMap<T, R>(this, mapper));
}

ObservableMap等操作符一般繼承AbstractObservableWithUpstream,這種Observable持有了上游引用

public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
    final Function<? super T, ? extends U> function;
    public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends U> function) {
        super(source);
        this.function = function;
    }
    
    @Override
    public void subscribeActual(Observer<? super U> downStream) {
        // 合并下游和mapper蛹磺,創(chuàng)建新的Observer:MapObserver
        source.subscribe(new MapObserver<T, U>(downStream, function));
    }
    
    static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
        final Function<? super T, ? extends U> mapper;
        MapObserver(Observer<? super U> downStream, Function<? super T, ? extends U> mapper) {
            super(downStream);
            this.mapper = mapper;
        }
        
        // 重新實(shí)現(xiàn)onNext
        @Override
        public void onNext(T t) {
            if (done) {
                return;
            }
            if (sourceMode != NONE) {
                downStream.onNext(null);
                return;
            }
            U v;
            try {
                // 檢查mapper的輸出結(jié)果粟瞬,合法則賦值給
                v = ObjectHelper.requireNonNull(
                    mapper.apply(t), "The mapper function returned a null value.");
            } catch (Throwable ex) {
                fail(ex);
                return;
            }
            // 下游onNext傳入經(jīng)過mapper的結(jié)果
            downStream.onNext(v);
        }
    }

對(duì)于操作符而言

  • 自上而下
    • 裝飾器模式。操作符創(chuàng)建了一個(gè)新的 Observable
    • 該 Observable 重寫了subscribeActual方法
    • subscribeActual 的實(shí)現(xiàn)是讓所持有的上游source調(diào)用subscribe方法去訂閱 被重新裝飾的下游
  • 自下而上
    • 裝飾器模式萤捆。操作符使下游Observer被裝飾后形成新的Observer
    • 重寫了Observer的幾個(gè)方法裙品,向下傳遞經(jīng)過操作符處理后的數(shù)值結(jié)果

自定義操作符

// compose()操作符,自上而下俗或,封裝Observable
public interface ObservableTransformer<Upstream, Downstream> {
    ObservableSource<Downstream> apply(Observable<Upstream> upstream);
}
// lift()操作符市怎,自下而上,封裝Observer
public interface ObservableOperator<Downstream, Upstream> {
    Observer<? super Upstream> apply(@NonNull Observer<? super Downstream> observer) throws Exception;
}
  • lift
    • 創(chuàng)建并返回新的Observable辛慰,即ObservableLift
    • 訂閱發(fā)生在operator.apply(s)区匠,即得到新的Observable之后
    • 意圖是封裝一個(gè)操作符,類似于create
  • compose
    • 意圖是封裝一系列操作符昆雀,方便復(fù)用

四辱志、線程調(diào)度

舉例 .subscribeOn(AndroidSchedulers.mainThread())

public final Observable<T> subscribeOn(Scheduler scheduler) {
    ObjectHelper.requireNonNull(scheduler, "scheduler is null");
    return RxJavaPlugins.onAssembly(
        // 返回Observable蝠筑,傳入上游this和調(diào)度器scheduler
        new ObservableSubscribeOn<T>(this, scheduler));
}
    
public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
    final Scheduler scheduler;
    public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
        super(source);
        this.scheduler = scheduler;
    }
    @Override
    public void subscribeActual(final Observer<? super T> downStream) {
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(downStream);
        // SubscribeOnObserver實(shí)現(xiàn)了Disposable狞膘,將其傳遞給下游的onSubscribe
        downStream.onSubscribe(parent);
        // 把source的訂閱放在Runnable中,由scheduler調(diào)度
        parent.setDisposable(scheduler.scheduleDirect(new Runnable() {
            @Override
            public void run() {
                source.subscribe(parent);
            }
        }));
    }
}

講解

  • subscribeOn 線程調(diào)度即發(fā)生在subscribe時(shí)
  • subscribeOn 只能生效一次
    • 因?yàn)橥暾挠嗛嗊^程是自下而上訂閱什乙,數(shù)據(jù)源發(fā)射事件在自上而下傳遞挽封,所以真正發(fā)射事件所在的線程,是有最接近上游的一次subscribeOn來決定的臣镣,其他的都會(huì)被覆蓋
    • 從代碼上看辅愿,就是經(jīng)歷了n個(gè)線程的傳遞后,把source.subscribe(parent)放在了最上游的那個(gè)線程中去發(fā)射忆某,如果沒有observeOn影響点待,整個(gè)事件流都會(huì)在那個(gè)線程完成
    • 為什么flatMap能改變subscribeOn的這種特性?因?yàn)閒latMap等操作符創(chuàng)建了新的Observable弃舒,而不是單純傳遞上下游

對(duì)于調(diào)度器 AndroidSchedulers.mainThread()癞埠。其實(shí)際上是 new HandlerScheduler(new Handler(Looper.getMainLooper()))

從上面看出状原,實(shí)現(xiàn)線程調(diào)度的是scheduler.scheduleDirect方法

@Override
public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
    if (run == null) throw new NullPointerException("run == null");
    if (unit == null) throw new NullPointerException("unit == null");

    run = RxJavaPlugins.onSchedule(run);
    // ScheduledRunnable 實(shí)現(xiàn)了Disposable接口
    ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
    // 大家熟悉的handler.postDelayed
    handler.postDelayed(scheduled, Math.max(0L, unit.toMillis(delay)));
    return scheduled;
}

// 一般的Schedule都是用createWorker().schedule(scheduled, delay, unit)的方式去實(shí)現(xiàn)線程調(diào)度
// Worker實(shí)現(xiàn)相應(yīng)的disposable接口,便于取消訂閱時(shí)停止執(zhí)行尚未執(zhí)行的Runnable苗踪,并裝飾相應(yīng)的hook颠区。
// 這個(gè)類的實(shí)現(xiàn)比較有點(diǎn)劃水,可能因?yàn)槭侵骶€程通铲,所以不擔(dān)心泄露
@Override
public Worker createWorker() {
    return new HandlerWorker(handler);
}

private static final class HandlerWorker extends Worker {
    private final Handler handler;
    private volatile boolean disposed;

    HandlerWorker(Handler handler) {
        this.handler = handler;
    }

    @Override
    public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
        if (run == null) throw new NullPointerException("run == null");
        if (unit == null) throw new NullPointerException("unit == null");

        if (disposed) {
            return Disposables.disposed();
        }

        run = RxJavaPlugins.onSchedule(run);

        ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);

        Message message = Message.obtain(handler, scheduled);
        message.obj = this; // Used as token for batch disposal of this worker's runnables.

        handler.sendMessageDelayed(message, Math.max(0L, unit.toMillis(delay)));

        // Re-check disposed state for removing in case we were racing a call to dispose().
        if (disposed) {
            handler.removeCallbacks(scheduled);
            return Disposables.disposed();
        }

        return scheduled;
    }

    @Override
    public void dispose() {
        disposed = true;
        handler.removeCallbacksAndMessages(this /* token */);
    }

    @Override
    public boolean isDisposed() {
        return disposed;
    }
}

舉例 .observeOn(AndroidSchedulers.mainThread())

public final Observable<T> observeOn(Scheduler scheduler) {
    return observeOn(scheduler, false, bufferSize());
}
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
    ObjectHelper.requireNonNull(scheduler, "scheduler is null");
    ObjectHelper.verifyPositive(bufferSize, "bufferSize");
    return RxJavaPlugins.onAssembly(
        new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
}
// 以上代碼都很熟悉了毕莱,暫不贅述

public final class ObservableObserveOn<T> extends AbstractObservableWithUpstream<T, T> {
    final Scheduler scheduler;
    final boolean delayError;
    final int bufferSize;
    public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
        super(source);
        this.scheduler = scheduler; // 線程調(diào)度器
        this.delayError = delayError; // 出現(xiàn)錯(cuò)誤是否立刻中斷線程
        this.bufferSize = bufferSize;   //緩沖區(qū)大小,默認(rèn)128
    }
    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        if (scheduler instanceof TrampolineScheduler) {
            // 默認(rèn)線程則不做線程調(diào)度颅夺,直接在當(dāng)前線程中調(diào)用
            source.subscribe(observer);
        } else {
            Scheduler.Worker w = scheduler.createWorker();
            source.subscribe(
                // 用Worker朋截、和相關(guān)參數(shù)裝飾observer,得到新的Observer注入上游
                new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
        }
    }
  
  static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T> implements Observer<T>, Runnable {
    //省略部分代碼
    @Override
        public void onSubscribe(Disposable s) {
            if (DisposableHelper.validate(this.s, s)) {
                this.s = s;
                //省略部分代碼吧黄,創(chuàng)建緩沖隊(duì)列
                queue = new SpscLinkedArrayQueue<T>(bufferSize);
                actual.onSubscribe(this);
            }
    }
    
    @Override
    public void onNext(T t) {
      if (done) {
        return;
      }
      if (sourceMode != QueueDisposable.ASYNC) {
        queue.offer(t); //上游的數(shù)據(jù)全部先入隊(duì)列
      }
      //執(zhí)行調(diào)度
      schedule();
    }
    
    void schedule() {
      if (getAndIncrement() == 0) {
        // 隊(duì)列如果已經(jīng)空了质和,則再次調(diào)度
        worker.schedule(this);
      }
    }
    
    @Override
    public void run() {
        // Fused 熔斷機(jī)制,默認(rèn)false
        if (outputFused) {
            drainFused();
        } else {
            drainNormal();
        }
    }
    
    //該函數(shù)在Runnable所在的線程執(zhí)行稚字,從緩沖隊(duì)列里拿出事件饲宿,向下游發(fā)射
    void drainNormal() {
        int missed = 1;
        final SimpleQueue<T> q = queue;
        final Observer<? super T> a = actual;
        for (;;) {
            // 如果設(shè)置了errorDelay,則不管隊(duì)列是否為空胆描,發(fā)生了錯(cuò)誤都會(huì)中斷發(fā)射瘫想,并調(diào)用observer的onError
            if (checkTerminated(done, q.isEmpty(), a)) {
                return;
            }
            for (;;) {
                boolean d = done;
                T v;
                try {
                    v = q.poll();   //隊(duì)列中取數(shù)據(jù)
                } catch (Throwable ex) {
                    Exceptions.throwIfFatal(ex);
                    s.dispose();
                    q.clear();
                    a.onError(ex);
                    worker.dispose();
                    return;
                }
                boolean empty = v == null;
                if (checkTerminated(d, empty, a)) {
                    return;
                }
                if (empty) {
                    break;
                }
                // 向下游發(fā)射數(shù)據(jù)
                a.onNext(v);
            }
            missed = addAndGet(-missed);
            if (missed == 0) {
                break;
            }
        }
    }
  }
}

講解

  • 訂閱發(fā)生在調(diào)度前,說明線程調(diào)度不影響訂閱過程
  • ObserveOnObserver持有下游observer和調(diào)度器,并實(shí)現(xiàn)Runnable接口
  • 訂閱時(shí)(onSubscribe)會(huì)創(chuàng)建一個(gè)緩沖隊(duì)列挚冤,當(dāng)上游數(shù)據(jù)到來先放在隊(duì)列威始,接著在調(diào)度線程中取出并發(fā)射到下游
  • 由上看observeOn可以多次生效

五、背壓

  • 默認(rèn)策略判斷是否觸發(fā)背壓的因素:
    • 同步場景中车吹,有發(fā)射數(shù)是否超出響應(yīng)式拉取值 request 決定
    • 異步場景中,由是否超出緩沖池 queue 的承受能力決定醋闭。需要下游的request方法拉取queue的數(shù)據(jù)
    • observeOn允許我們設(shè)置緩沖隊(duì)列的容量大小
    • 在onSubscribe(Subscription s)回調(diào)提供的s可以調(diào)用request方法來增加拉取數(shù)窄驹;如果不重寫,默認(rèn)執(zhí)行s.request(Long.MAX_VALUE)
  • 背壓策略
    • ERROR 觸發(fā)背壓直接拋異常 MissingBackpressureException
    • BUFFER: queue無限大证逻,知道OOM
    • DROP: 超載則拋棄之后的數(shù)據(jù)乐埠,不拋異常
    • LATEST:超載后拋棄之后數(shù)據(jù),且是專用有一個(gè)額外空間保留當(dāng)前最新一次數(shù)據(jù)

舉例 onBackpressureDrop 方法

public final class FlowableOnBackpressureDrop<T> extends AbstractFlowableWithUpstream<T, T> implements Consumer<T> {
    final Consumer<? super T> onDrop;
  
    @Override
    protected void subscribeActual(Subscriber<? super T> s) {
        this.source.subscribe(new BackpressureDropSubscriber<T>(s, onDrop));
    }
  
    static final class BackpressureDropSubscriber<T> extends AtomicLong implements FlowableSubscriber<T>, Subscription {
      @Override
        public void onNext(T t) {
            if (done) {
                return;
            }
            long r = get();
            if (r != 0L) {
                actual.onNext(t);
                BackpressureHelper.produced(this, 1);
            } else {
                try {
                    onDrop.accept(t);
                } catch (Throwable e) {
                    Exceptions.throwIfFatal(e);
                    cancel();
                    onError(e);
                }
            }
        }
      
      @Override
        public void request(long n) {
            if (SubscriptionHelper.validate(n)) {
                BackpressureHelper.add(this, n);
            }
        }
    }
}

BackpressureDropSubscriber繼承了AtomicLong囚企,實(shí)現(xiàn)了Subscriber丈咐。只有判斷自身不為0時(shí)才會(huì)向下游發(fā)射元素,否則將被拋棄龙宏。這個(gè)數(shù)值的計(jì)算在BackpressureHelper中計(jì)算

// 每執(zhí)行一次onNext棵逊,當(dāng)前值減一
public static long produced(AtomicLong requested, long n) {
    for (;;) {
        long current = requested.get();
        if (current == Long.MAX_VALUE) {
            return Long.MAX_VALUE;
        }
        long update = current - n;
        if (update < 0L) {
            RxJavaPlugins.onError(new IllegalStateException("More produced than requested: " + update));
            update = 0L;
        }
        if (requested.compareAndSet(current, update)) {
            return update;
        }
    }
}

// request調(diào)用時(shí),值會(huì)加到當(dāng)前值上银酗。
public static long add(AtomicLong requested, long n) {
    for (;;) {
        long r = requested.get();
        if (r == Long.MAX_VALUE) {
            return Long.MAX_VALUE;
        }
        long u = addCap(r, n);
        if (requested.compareAndSet(r, u)) {
            return r;
        }
    }
}

這樣辆影,就實(shí)現(xiàn)了發(fā)生背壓則拋棄新的值

onBackpressureLatest的實(shí)現(xiàn)和前者類似掩浙,只是多了一個(gè)對(duì)象來存儲(chǔ)最新一次的值:

final AtomicReference<T> current = new AtomicReference<T>();
@Override
public void onNext(T t) {
     current.lazySet(t);
     drain();
}

六、Subject

Subject 繼承Observable秸歧,實(shí)現(xiàn)Observer厨姚。具有自攻自受的特性,相當(dāng)于是一個(gè)中間層

一般Observable觀察的對(duì)象往往是靜態(tài)的键菱,如一個(gè)常量谬墙、一個(gè)文件。有確定的開頭和結(jié)尾

Subject是動(dòng)態(tài)的经备,監(jiān)聽沒有被中斷的時(shí)候拭抬,可以有不確定的事件傳來

  • Observable特性
    • 訂閱:如果下游可用,則加入到訂閱隊(duì)列中
@Override
public void subscribeActual(Observer<? super T> t) {
    PublishDisposable<T> ps = new PublishDisposable<T>(t, this);
    t.onSubscribe(ps);
    if (add(ps)) {
        // if cancellation happened while a successful add, the remove() didn't work
        // so we need to do it again
        if (ps.isDisposed()) {
            remove(ps);
        }
    } else {
        Throwable ex = error;
        if (ex != null) {
            t.onError(ex);
        } else {
            t.onComplete();
        }
    }
}

boolean add(PublishDisposable<T> ps) {
    for (;;) {
        PublishDisposable<T>[] a = subscribers.get();
        if (a == TERMINATED) {
            return false;
        }

        int n = a.length;
        @SuppressWarnings("unchecked")
        PublishDisposable<T>[] b = new PublishDisposable[n + 1];
        System.arraycopy(a, 0, b, 0, n);
        b[n] = ps;

        if (subscribers.compareAndSet(a, b)) {
            return true;
        }
    }
}
  • Observer 特性
    • 如果本身可用侵蒙,切存在可用的觀察者造虎,則向下傳遞事件
// 但舉例onNext方法
@Override
public void onNext(T t) {
    if (subscribers.get() == TERMINATED) {
        return;
    }
    if (t == null) {
        onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
        return;
    }
    // 遍歷下游觀察者隊(duì)列,并逐個(gè)調(diào)用onNext
    for (PublishDisposable<T> s : subscribers.get()) {
        s.onNext(t);
    }
}

七纷闺、性能問題

.map(x -> x + 1) 和 .flatMap(x -> Observable.just(x + 1))有什么區(qū)別
  • map和flatMap的區(qū)別
    • map沒有創(chuàng)建新的Observable算凿,flatMap創(chuàng)建了新的Observable,相當(dāng)于創(chuàng)建了新的流
    • map在上游和下游之間仍是線性的犁功。flatMap已經(jīng)上升了一個(gè)階氓轰,即每一個(gè)元素進(jìn)來,都會(huì)變成一個(gè)新的source浸卦,而下游也會(huì)變成一個(gè)新的observer

循環(huán)的靜態(tài)Observable和動(dòng)態(tài)的Subject之間的區(qū)別

void callback(int item) {
    Observable
        .<Integer>create(emitter -> {
            emitter.onNext(item);
        })
        .map(i -> i + 1)
        .subscribe(i -> Log.d(TAG, "i = " + i));
}
Subject<Integer> mSubject = PublishSubject.create();

void init() {
    mSuject
        .map(i -> i + 1)
        .subscribe(i -> Log.d(TAG, "i = " + i));
}

void callback(int item) {
    mSuject.onNext(item);
}
  • 循環(huán)創(chuàng)建的事件流署鸡,所有操作符都涉及了new對(duì)象,如果callback被頻繁調(diào)用限嫌,則會(huì)產(chǎn)生十分多的臨時(shí)對(duì)象造成內(nèi)存抖動(dòng)

八靴庆、Rx擴(kuò)展

RxBinding

final class ViewClickObservable extends Observable<Object> {
    private final View view;
    ViewClickObservable(View view) {
        this.view = view;
    }
    @Override protected void subscribeActual(Observer<? super Object> observer) {
        if (!checkMainThread(observer)) {
            return;
        }
        //訂閱階段執(zhí)行的邏輯,創(chuàng)建listener綁定view
        Listener listener = new Listener(view, observer);
        observer.onSubscribe(listener);
        view.setOnClickListener(listener);
    }
    static final class Listener extends MainThreadDisposable implements OnClickListener {
        private final View view;
        private final Observer<? super Object> observer;
        Listener(View view, Observer<? super Object> observer) {
            this.view = view;
            this.observer = observer;
        }
        @Override public void onClick(View v) {
            if (!isDisposed()) {
                //發(fā)生點(diǎn)擊時(shí)傳遞事件給下游
                observer.onNext(Notification.INSTANCE);
            }
        }
        @Override protected void onDispose() {
            view.setOnClickListener(null);
        }
    }
}

RxPermissions

public <T> ObservableTransformer<T, Boolean> ensure(final String... permissions) {
    return new ObservableTransformer<T, Boolean>() {
        @Override
        public ObservableSource<Boolean> apply(Observable<T> o) {
            return request(o, permissions)
                    // 轉(zhuǎn)換 Observable<Permission> 為 Observable<Boolean>
                    .buffer(permissions.length)
                    .flatMap(new Function<List<Permission>, ObservableSource<Boolean>>() {
                        @Override
                        public ObservableSource<Boolean> apply(List<Permission> permissions) {
                            if (permissions.isEmpty()) {
                                // Occurs during orientation change, when the subject receives onComplete.
                                // In that case we don't want to propagate that empty list to the
                                // subscriber, only the onComplete.
                                return Observable.empty();
                            }
                            // Return true if all permissions are granted.
                            for (Permission p : permissions) {
                                if (!p.granted) {
                                    return Observable.just(false);
                                }
                            }
                            return Observable.just(true);
                        }
                    });
        }
    };
}
注:本文為了節(jié)省篇幅怒医,對(duì)源碼進(jìn)行了一定程度的刪減炉抒,僅供參考。

參考文章

最后編輯于
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
  • 序言:七十年代末裆熙,一起剝皮案震驚了整個(gè)濱河市端礼,隨后出現(xiàn)的幾起案子禽笑,更是在濱河造成了極大的恐慌入录,老刑警劉巖,帶你破解...
    沈念sama閱讀 218,546評(píng)論 6 507
  • 序言:濱河連續(xù)發(fā)生了三起死亡事件佳镜,死亡現(xiàn)場離奇詭異僚稿,居然都是意外死亡,警方通過查閱死者的電腦和手機(jī)蟀伸,發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 93,224評(píng)論 3 395
  • 文/潘曉璐 我一進(jìn)店門蚀同,熙熙樓的掌柜王于貴愁眉苦臉地迎上來缅刽,“玉大人,你說我怎么就攤上這事蠢络∷ッ停” “怎么了?”我有些...
    開封第一講書人閱讀 164,911評(píng)論 0 354
  • 文/不壞的土叔 我叫張陵刹孔,是天一觀的道長啡省。 經(jīng)常有香客問我,道長髓霞,這世上最難降的妖魔是什么卦睹? 我笑而不...
    開封第一講書人閱讀 58,737評(píng)論 1 294
  • 正文 為了忘掉前任,我火速辦了婚禮方库,結(jié)果婚禮上结序,老公的妹妹穿的比我還像新娘。我一直安慰自己纵潦,他們只是感情好徐鹤,可當(dāng)我...
    茶點(diǎn)故事閱讀 67,753評(píng)論 6 392
  • 文/花漫 我一把揭開白布。 她就那樣靜靜地躺著邀层,像睡著了一般凳干。 火紅的嫁衣襯著肌膚如雪。 梳的紋絲不亂的頭發(fā)上被济,一...
    開封第一講書人閱讀 51,598評(píng)論 1 305
  • 那天救赐,我揣著相機(jī)與錄音,去河邊找鬼只磷。 笑死经磅,一個(gè)胖子當(dāng)著我的面吹牛,可吹牛的內(nèi)容都是我干的钮追。 我是一名探鬼主播预厌,決...
    沈念sama閱讀 40,338評(píng)論 3 418
  • 文/蒼蘭香墨 我猛地睜開眼,長吁一口氣:“原來是場噩夢啊……” “哼元媚!你這毒婦竟也來了轧叽?” 一聲冷哼從身側(cè)響起,我...
    開封第一講書人閱讀 39,249評(píng)論 0 276
  • 序言:老撾萬榮一對(duì)情侶失蹤刊棕,失蹤者是張志新(化名)和其女友劉穎炭晒,沒想到半個(gè)月后,有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體甥角,經(jīng)...
    沈念sama閱讀 45,696評(píng)論 1 314
  • 正文 獨(dú)居荒郊野嶺守林人離奇死亡网严,尸身上長有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點(diǎn)故事閱讀 37,888評(píng)論 3 336
  • 正文 我和宋清朗相戀三年,在試婚紗的時(shí)候發(fā)現(xiàn)自己被綠了嗤无。 大學(xué)時(shí)的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片震束。...
    茶點(diǎn)故事閱讀 40,013評(píng)論 1 348
  • 序言:一個(gè)原本活蹦亂跳的男人離奇死亡怜庸,死狀恐怖,靈堂內(nèi)的尸體忽然破棺而出垢村,到底是詐尸還是另有隱情割疾,我是刑警寧澤,帶...
    沈念sama閱讀 35,731評(píng)論 5 346
  • 正文 年R本政府宣布嘉栓,位于F島的核電站杈曲,受9級(jí)特大地震影響,放射性物質(zhì)發(fā)生泄漏胸懈。R本人自食惡果不足惜担扑,卻給世界環(huán)境...
    茶點(diǎn)故事閱讀 41,348評(píng)論 3 330
  • 文/蒙蒙 一、第九天 我趴在偏房一處隱蔽的房頂上張望趣钱。 院中可真熱鬧涌献,春花似錦、人聲如沸首有。這莊子的主人今日做“春日...
    開封第一講書人閱讀 31,929評(píng)論 0 22
  • 文/蒼蘭香墨 我抬頭看了看天上的太陽井联。三九已至卜壕,卻和暖如春,著一層夾襖步出監(jiān)牢的瞬間烙常,已是汗流浹背轴捎。 一陣腳步聲響...
    開封第一講書人閱讀 33,048評(píng)論 1 270
  • 我被黑心中介騙來泰國打工, 沒想到剛下飛機(jī)就差點(diǎn)兒被人妖公主榨干…… 1. 我叫王不留蚕脏,地道東北人侦副。 一個(gè)月前我還...
    沈念sama閱讀 48,203評(píng)論 3 370
  • 正文 我出身青樓,卻偏偏與公主長得像驼鞭,于是被迫代替她去往敵國和親秦驯。 傳聞我的和親對(duì)象是個(gè)殘疾皇子,可洞房花燭夜當(dāng)晚...
    茶點(diǎn)故事閱讀 44,960評(píng)論 2 355

推薦閱讀更多精彩內(nèi)容

  • 引入依賴: implementation 'io.reactivex.rxjava2:rxandroid:2.0....
    為夢想戰(zhàn)斗閱讀 1,304評(píng)論 0 0
  • 一挣棕、RxJava操作符概述 RxJava中的操作符就是為了提供函數(shù)式的特性译隘,函數(shù)式最大的好處就是使得數(shù)據(jù)處理簡潔易...
    BrotherChen閱讀 1,615評(píng)論 0 10
  • 一、RxJava操作符概述 RxJava中的操作符就是為了提供函數(shù)式的特性洛心,函數(shù)式最大的好處就是使得數(shù)據(jù)處理簡潔易...
    測天測地測空氣閱讀 636評(píng)論 0 1
  • 轉(zhuǎn)載自:https://xiaobailong24.me/2017/03/18/Android-RxJava2.x...
    Young1657閱讀 2,022評(píng)論 1 9
  • 我從去年開始使用 RxJava 固耘,到現(xiàn)在一年多了。今年加入了 Flipboard 后皂甘,看到 Flipboard 的...
    Jason_andy閱讀 5,473評(píng)論 7 62