博客
关于我
强烈建议你试试无所不能的chatGPT，快点击我
Rxjava
阅读量：6911 次
发布时间：2019-06-27

本文共 21563 字，大约阅读时间需要 71 分钟。

RxJava

RxJava是基于Java虚拟机上的响应式扩展库，它通过使用可观察的序列将异步和基于事件的程序组合起来。 与此同时，它扩展了观察者模式来支持数据/事件序列，并且添加了操作符，这些操作符允许你声明性地组合序列，同时抽象出要关注的问题：比如低级线程、同步、线程安全和并发数据结构等。

从RxJava的官方定义来看，我们如果要想真正地理解RxJava，就必须对它以下两个部分进行深入的分析：

1、订阅流程。

2、线程切换。

RxJava的订阅流程

Observable.create(new ObservableOnSubscribe
    
     () {    @Override    public void subscribe(ObservableEmitter
     
      emitter) throws Exception {        emitter.onNext("1");        emitter.onNext("2");        emitter.onNext("3");        emitter.onComplete();    }}).subscribe(new Observer
      
       () {    @Override    public void onSubscribe(Disposable d) {        Log.d(TAG, "onSubscribe");    }    @Override    public void onNext(String s) {        Log.d(TAG, "onNext : " + s);    }    @Override    public void onError(Throwable e) {        Log.d(TAG, "onError : " + e.toString());    }    @Override    public void onComplete() {        Log.d(TAG, "onComplete");    }});复制代码

可以看到，这里首先创建了一个被观察者，然后创建一个观察者订阅了这个被观察者，因此下面分两个部分对RxJava的订阅流程进行分析：

1、创建被观察者过程。

2、订阅过程。

创建被观察者过程

上面使用了Observable类的create()方法创建了一个被观察者，看看里面做了什么。

Observable#create()

public static 
    
      Observable
     
       create(ObservableOnSubscribe
      
        source) {    ObjectHelper.requireNonNull(source, "source is null");    return RxJavaPlugins.onAssembly(new ObservableCreate
       
        (source));}public static 
        
          T requireNonNull(T object, String message) {    if (object == null) {         throw new NullPointerException(message);    }    return object; }public static 
         
           Observable
          
            onAssembly(@NonNull Observable
           
             source) { Function
             f = onObservableAssembly; if (f != null) { return apply(f, source); } return source;}复制代码

在Observable的create()里面实际上是创建了一个新的ObservableCreate对象，同时，把我们定义好的ObservableOnSubscribe对象传入了ObservableCreate对象中，最后调用了RxJavaPlugins.onAssembly()方法。接下来看看这个ObservableCreate是干什么的。

ObservableCreate

public final class ObservableCreate
    
      extends Observable
     
       {    final ObservableOnSubscribe
      
        source;    public ObservableCreate(ObservableOnSubscribe
       
         source) {        this.source = source;    }    ...}复制代码

这里仅仅是把ObservableOnSubscribe这个对象保存在ObservableCreate中了。然后看看RxJavaPlugins.onAssembly()这个方法的处理。

RxJavaPlugins#onAssembly()

public static 
    
      Observable
     
       onAssembly(@NonNull Observable
      
        source) {    // 应用hook函数的一些处理，一般用到不到    ...    return source;}复制代码

最终仅仅是把我们的ObservableCreate给返回了。

Observable.create()方法仅仅是先将我们自定义的ObservableOnSubscribe对象重新包装成了一个ObservableCreate对象。

接着是创建Observer，这比较简单只是单纯创建一个接口对象而已。

public interface Observer
    
      {    void onSubscribe(@NonNull Disposable d);    void onNext(@NonNull T t);    void onError(@NonNull Throwable e);        void onComplete();}复制代码

订阅过程

Observable#subscribe()

public final void subscribe(Observer
     observer) {    ObjectHelper.requireNonNull(observer, "observer is null");    try {        observer = RxJavaPlugins.onSubscribe(this, observer);        ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");        subscribeActual(observer);    } catch (NullPointerException e) { // NOPMD        throw e;    } catch (Throwable e) {        Exceptions.throwIfFatal(e);        RxJavaPlugins.onError(e);        NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");        npe.initCause(e);        throw npe;    }}//ObjectHelper.requireNonNull()方法public static 
    
      T requireNonNull(T object, String message) {    if (object == null) {         throw new NullPointerException(message);    }   return object;}//RxJavaPlugins.onSubscribe()方法public static 
     
       Observer
       onSubscribe(@NonNull Observable
      
        source, @NonNull Observer
        observer) {    BiFunction
        f = onObservableSubscribe;    if (f != null) {        return apply(f, source, observer);    }    return observer;}复制代码

在Observable的subscribe()方法内部首先调用了RxJavaPlugins的onSubscribe()方法。

RxJavaPlugins#onSubscribe()

public static 
    
      Observer
      onSubscribe(@NonNull Observable
     
       source, @NonNull Observer
       observer) {    // 应用hook函数的一些处理，一般用到不到    ...    return observer;}复制代码

这里仅仅是将observer返回了。接着来分析下subscribeActual()方法，

Observable#subscribeActual()

protected abstract void subscribeActual(Observer
     observer);复制代码

这是一个抽象的方法，很明显，它对应的具体实现类就是我们在第一步创建的ObservableCreate类，接下来看到ObservableCreate的subscribeActual()方法。

ObservableCreate#subscribeActual()

@Overrideprotected void subscribeActual(Observer
     observer) {    // 1    CreateEmitter
    
      parent = new CreateEmitter
     
      (observer);    // 2    observer.onSubscribe(parent);    try {        // 3        source.subscribe(parent);    } catch (Throwable ex) {        Exceptions.throwIfFatal(ex);        parent.onError(ex);    }}复制代码

首先新创建了一个CreateEmitter对象，同时传入了我们自定义的observer对象进去。然后回到observer的onSubscribe()方法，而source就是我们之前创建ObservableCreate传入的ObservableOnSubscribe对象。

CreateEmitter

static final class CreateEmitter
    
     extends AtomicReference
     
      implements ObservableEmitter
      
       , Disposable {    ...    final Observer
        observer;    CreateEmitter(Observer
        observer) {        this.observer = observer;    }    ...}复制代码

从上面可以看出，CreateEmitter通过继承了Java并发包中的原子引用类AtomicReference保证了事件流切断状态Dispose的一致性，并实现了ObservableEmitter接口和Disposable接口，接着我们分析下注释2处的observer.onSubscribe(parent)，这个onSubscribe回调的含义其实就是告诉观察者已经成功订阅了被观察者。再看到注释3处的source.subscribe(parent)这行代码，这里的source其实是ObservableOnSubscribe对象，我们看到ObservableOnSubscribe的subscribe()方法。

ObservableOnSubscribe#subscribe()

Observable observable = Observable.create(new ObservableOnSubscribe
    
     () {    @Override    public void subscribe(ObservableEmitter
     
       emitter) throws Exception {        emitter.onNext("1");        emitter.onNext("2");        emitter.onNext("3");        emitter.onComplete();    }});复制代码

这里面使用到了ObservableEmitter的onNext()方法将事件流发送出去，最后调用了onComplete()方法完成了订阅过程。ObservableEmitter是一个抽象类，实现类就是我们传入的CreateEmitter对象，接下来我们看看CreateEmitter的onNext()方法和onComplete()方法的处理。

CreateEmitter#onNext() && CreateEmitter#onComplete()

static final class CreateEmitter
    
     extends AtomicReference
     
      implements ObservableEmitter
      
       , Disposable {...@Overridepublic void onNext(T t) {    ...    if (!isDisposed()) {        //调用观察者的onNext()        observer.onNext(t);    }}@Overridepublic void onComplete() {    if (!isDisposed()) {        try {            observer.onComplete();        } finally {            dispose();        }    }}...}复制代码

在CreateEmitter的onNext和onComplete方法中首先都要经过一个isDisposed的判断，作用就是看当前的事件流是否被切断（废弃）掉了，默认是不切断的，如果想要切断，可以调用Disposable的dispose()方法将此状态设置为切断（废弃）状态。我们继续看看这个isDisposed内部的处理。可以看到最终又回调了观察者的onNext()方法，把被观察者的数据传输给了观察者。

ObservableEmitter#isDisposed()

@Overridepublic boolean isDisposed() {    return DisposableHelper.isDisposed(get());}复制代码

注意到这里通过get()方法首先从ObservableEmitter的AtomicReference中拿到了保存的Disposable状态。然后交给了DisposableHelper进行判断处理。接下来看看DisposableHelper的处理。

DisposableHelper#isDisposed() && DisposableHelper#set()

public enum DisposableHelper implements Disposable {    DISPOSED;    public static boolean isDisposed(Disposable d) {        // 1        return d == DISPOSED;    }    public static boolean set(AtomicReference
    
      field, Disposable d) {        for (;;) {            Disposable current = field.get();            if (current == DISPOSED) {                if (d != null) {                    d.dispose();                }                return false;            }            // 2            if (field.compareAndSet(current, d)) {                if (current != null) {                    current.dispose();                }                return true;            }        }    }    ...    public static boolean dispose(AtomicReference
     
       field) {        Disposable current = field.get();        Disposable d = DISPOSED;        if (current != d) {            // ...            current = field.getAndSet(d);            if (current != d) {                if (current != null) {                    current.dispose();                }                return true;            }        }        return false;    }    ...}复制代码

DisposableHelper是一个枚举类，内部只有一个值即DISPOSED, 从上面的分析可知它就是用来标记事件流被切断（废弃）状态的。先看到注释2和注释3处的代码field.compareAndSet(current, d)和field.getAndSet(d)，这里使用了原子引用AtomicReference内部包装的CAS方法处理了标志Disposable的并发读写问题，最后看到注释3处，将我们传入的CreateEmitter这个原子引用类保存的Dispable状态和DisposableHelper内部的DISPOSED进行比较，如果相等，就证明数据流被切断了。为了更进一步理解Disposed的作用，再来看看CreateEmitter中剩余的关键方法。

CreateEmitter

@Overridepublic void onNext(T t) {    ...    // 1    if (!isDisposed()) {        observer.onNext(t);    }}@Overridepublic void onError(Throwable t) {    if (!tryOnError(t)) {        // 2        RxJavaPlugins.onError(t);    }}@Overridepublic boolean tryOnError(Throwable t) {    ...    // 3    if (!isDisposed()) {        try {            observer.onError(t);        } finally {            // 4            dispose();        }        return true;    }    return false;}@Overridepublic void onComplete() {    // 5    if (!isDisposed()) {        try {            observer.onComplete();        } finally {            // 6            dispose();        }    }}复制代码

在注释1、3、5处，onNext()和onError()、onComplete()方法首先都会判断事件流是否被切断的处理，如果事件流此时被切断了，那么onNext()和onComplete()则会退出方法体，不做处理，onError()则会执行到RxJavaPlugins.onError(t)这句代码，内部会直接抛出异常，导致崩溃。如果事件流没有被切断，那么在onError()和onComplete()内部最终会调用到注释4、6处的这句dispose()代码，将事件流进行切断，由此可知，onError()和onComplete()只能调用一个，如果先执行的是onComplete()，再调用onError()的话就会导致异常崩溃。

RxJava的线程切换

Observable.create(new ObservableOnSubscribe
    
     () {    @Override    public void subscribe(ObservableEmitter
     
      emitter) throws Exception {        emitter.onNext("1");        emitter.onNext("2");        emitter.onNext("3");        emitter.onComplete();    }})     .subscribeOn(Schedulers.io())    .observeOn(AndroidSchedulers.mainThread())    .subscribe(new Observer
      
       () {        @Override        public void onSubscribe(Disposable d) {            Log.d(TAG, "onSubscribe");        }        @Override        public void onNext(String s) {            Log.d(TAG, "onNext : " + s);        }        @Override        public void onError(Throwable e) {            Log.d(TAG, "onError : " +e.toString());        }        @Override        public void onComplete() {            Log.d(TAG, "onComplete");        }});复制代码

RxJava的线程切换主要分为subscribeOn()和observeOn()方法。

subscribeOn(Schedulers.io())

在Schedulers.io()方法中，我们需要先传入一个Scheduler调度类，这里是传入了一个调度到io子线程的调度类，我们看看这个Schedulers.io()方法内部是怎么构造这个调度器的。

我们传进去了一个Scheduler类，Scheduler是一个调度类，能够延时或周期性地去执行一个任务。

Schedulers#io()

static final Scheduler IO;...public static Scheduler io() {    // 1    return RxJavaPlugins.onIoScheduler(IO);}static {    ...    // 2    IO = RxJavaPlugins.initIoScheduler(new IOTask());}static final class IOTask implements Callable
    
      {    @Override    public Scheduler call() throws Exception {        // 3        return IoHolder.DEFAULT;    }}static final class IoHolder {    // 4    static final Scheduler DEFAULT = new IoScheduler();}复制代码

首先，在注释1处，同前面分析的订阅流程的处理一样，只是一个处理hook的逻辑，最终返回的还是传入的这个IO对象。再看到注释2处，在Schedulers的静态代码块中将IO对象进行了初始化，其实质就是新建了一个IOTask的静态内部类，在IOTask的call方法中，也就是注释3处，可以了解到使用了静态内部类的方式把创建的IOScheduler对象给返回出去了。绕了这么大圈子，Schedulers.io方法其实质就是返回了一个IOScheduler对象。

Observable#subscribeOn()

public final Observable
    
      subscribeOn(Scheduler scheduler) {    ...    return RxJavaPlugins.onAssembly(new ObservableSubscribeOn
     
      (this, scheduler));}复制代码

在subscribeOn()方法里面，又将ObservableCreate包装成了一个ObservableSubscribeOn对象。我们关注到ObservableSubscribeOn类。

ObservableSubscribeOn

public final class ObservableSubscribeOn
    
      extends AbstractObservableWithUpstream
     
       {    final Scheduler scheduler;    public ObservableSubscribeOn(ObservableSource
      
        source, Scheduler scheduler) {        // 1        super(source);        this.scheduler = scheduler;    }    @Override    public void subscribeActual(final Observer
        observer) {        // 2        final SubscribeOnObserver
       
         parent = new SubscribeOnObserver
        
         (observer);        // 3        observer.onSubscribe(parent);        // 4        parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));    }...}复制代码

首先，在注释1处，将传进来的source和scheduler保存起来。接着，等到实际订阅的时候，就会执行到这个subscribeActual方法，在注释2处，将我们自定义的Observer包装成了一个SubscribeOnObserver对象。在注释3处，通知观察者订阅了被观察者。在注释4处，内部先创建了一个SubscribeTask对象，来看看它的实现。

ObservableSubscribeOn#SubscribeTask

final class SubscribeTask implements Runnable {    private final SubscribeOnObserver
    
      parent;    SubscribeTask(SubscribeOnObserver
     
       parent) {        this.parent = parent;    }    @Override    public void run() {        source.subscribe(parent);    }}复制代码

SubscribeTask是ObservableSubscribeOn的内部类，它实质上就是一个任务类，在它的run方法中会执行到source.subscribe(parent)的订阅方法，这个source其实就是我们在ObservableSubscribeOn构造方法中传进来的ObservableCreate对象。接下来看看scheduler.scheduleDirect()内部的处理。

Scheduler#scheduleDirect()

public Disposable scheduleDirect(@NonNull Runnable run) {    return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);}public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {    // 1    final Worker w = createWorker();    // 2    final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);    // 3    DisposeTask task = new DisposeTask(decoratedRun, w);    // 4    w.schedule(task, delay, unit);    return task;}复制代码

这里最后会执行到上面这个scheduleDirect()重载方法。首先，在注释1处，会调用createWorker()方法创建一个工作者对象Worker，它是一个抽象类，这里的实现类就是IoScheduler，下面，我们看看IoScheduler类的createWorker()方法。

IOScheduler#createWorker()

final AtomicReference
    
      pool;...public IoScheduler(ThreadFactory threadFactory) {    this.threadFactory = threadFactory;    this.pool = new AtomicReference
     
      (NONE);    start();}...@Overridepublic Worker createWorker() {    // 1    return new EventLoopWorker(pool.get());}static final class EventLoopWorker extends Scheduler.Worker {    ...    EventLoopWorker(CachedWorkerPool pool) {        this.pool = pool;        this.tasks = new CompositeDisposable();        // 2        this.threadWorker = pool.get();    }}复制代码

首先，在注释1处调用了pool.get()这个方法，pool是一个CachedWorkerPool类型的原子引用对象，它的作用就是用于缓存工作者对象Worker的。然后，将得到的CachedWorkerPool传入新创建的EventLoopWorker对象中。重点关注一下注释2处，这里讲CachedWorkerPool缓存的threadWorker对象保存起来了。

下面，我们继续分析3.6处代码段的注释2处的代码，这里又是一个关于hook的封装处理，最终还是返回的当前的Runnable对象。在注释3处新建了一个切断任务DisposeTask将decoratedRun和w包装了起来。最后在注释4处调用了工作者的schedule()方法。下面我们来分析下它内部的处理。

IoScheduler#schedule()

@Overridepublic Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit){    ...    return threadWorker.scheduleActual(action,delayTime, unit, tasks);}复制代码

内部调用了threadWorker的scheduleActual()方法，实际上是调用到了父类NewThreadWorker的scheduleActual()方法，我们继续看看NewThreadWorker的scheduleActual()方法中做的事情。

NewThreadWorker#scheduleActual()

public NewThreadWorker(ThreadFactory threadFactory) {    executor = SchedulerPoolFactory.create(threadFactory);}@NonNullpublic ScheduledRunnable scheduleActual(final Runnable run, long delayTime, @NonNull TimeUnit unit, @Nullable DisposableContainer parent) {    Runnable decoratedRun = RxJavaPlugins.onSchedule(run);    // 1    ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);    if (parent != null) {        if (!parent.add(sr)) {            return sr;        }    }    Future
     f;    try {        // 2        if (delayTime <= 0) {            // 3            f = executor.submit((Callable)sr);        } else {            // 4            f = executor.schedule((Callable)sr, delayTime, unit);        }        sr.setFuture(f);    } catch (RejectedExecutionException ex) {        if (parent != null) {            parent.remove(sr);        }        RxJavaPlugins.onError(ex);    }    return sr;}复制代码

在NewThreadWorker的scheduleActual()方法的内部，在注释1处首先会新建一个ScheduledRunnable对象，将Runnable对象和parent包装起来了，这里parent是一个DisposableContainer对象，它实际的实现类是CompositeDisposable类，即一个保存所有事件流是否被切断状态的容器，它内部的实现是使用了RxJava自己定义的一个简单的OpenHashSet类ji进行存储。最后注释2处，判断是否设置了延迟时间，如果设置了，则调用线程池的submit()方法立即进行线程切换，否则，调用schedule()方法进行延时执行线程切换。

为什么多次执行subscribeOn()，只有第一次有效？

从上面的分析，我们可以很容易了解到被观察者被订阅时是从最外面的一层（ObservableSubscribeOn）通知到里面的一层（ObservableOnSubscribe），当连续执行了到多次subscribeOn()的时候，其实就是先执行倒数第一次的subscribeOn()方法，直到最后一次执行的subscribeOn()方法肯定会覆盖前面的线程切换。

observeOn(AndroidSchedulers.mainThread())

public final Observable
    
      observeOn(Scheduler scheduler) {    return observeOn(scheduler, false, bufferSize());}public final Observable
     
       observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {    ....    return RxJavaPlugins.onAssembly(new ObservableObserveOn
      
       (this, scheduler, delayError, bufferSize));}复制代码

可以看到，observeOn()方法内部最终也是返回了一个ObservableObserveOn对象，我们直接来看看ObservableObserveOn的subscribeActual()方法。

ObservableObserveOn#subscribeActual()

@Overrideprotected void subscribeActual(Observer
     observer) {    // 1    if (scheduler instanceof TrampolineScheduler) {        // 2        source.subscribe(observer);    } else {        // 3        Scheduler.Worker w = scheduler.createWorker();        // 4        source.subscribe(new ObserveOnObserver
    
     (observer, w, delayError, bufferSize));    }}复制代码

首先，在注释1处，判断指定的调度器是不是TrampolineScheduler，这是一个不进行线程切换，立即执行当前代码的调度器，。如果是，则会直接调用调用ObservableSubscribeOn的subscribe()方法，如果不是，则会在注释3处创建一个工作者对象。然后在注释4处创建一个新的ObserveOnObserver将SubscribeOnobserver对象包装起来，ObserveOnObserver是执行观察者的线程，并传入ObservableSubscribeOn的subscribe()方法进行订阅。接下来看看ObserveOnObserver类的重点方法。

ObserveOnObserver

@Overridepublic void onNext(T t) {    ...    if (sourceMode != QueueDisposable.ASYNC) {        // 1        queue.offer(t);    }    schedule();}@Overridepublic void onError(Throwable t) {    ...    schedule();}@Overridepublic void onComplete() {    ...    schedule();}复制代码

去除非主线逻辑的代码，在ObserveOnObserver的onNext()和onError()、onComplete()方法中最后都会调用到schedule()方法。接着看schedule()方法，其中onNext()还会把消息存放到队列中。

ObserveOnObserver#schedule()

void schedule() {    if (getAndIncrement() == 0) {        worker.schedule(this);    }}复制代码

这里使用了worker进行调度ObserveOnObserver这个实现了Runnable的任务。worker就是在AndroidSchedulers.mainThread()中创建的，内部其实就是使用Handler进行线程切换的，此处不再赘述了。接着看ObserveOnObserver的run()方法。

这个对应这个HandlerScheduler这个类，所以createWorker()对应着：

private static final class MainHolder {        static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()));}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, unit.toMillis(delay));            if (disposed) {                handler.removeCallbacks(scheduled);                return Disposables.disposed();            }            return scheduled;        }}复制代码

在next()方法里，运用android自带的Handler消息机制，通过把方法包裹在Message里，同通过handler.sendMessageDelayed()发送消息，就会在ui线程里回调Next()方法，从而实现从子线程切换到android主线程的操作。

ObserveOnObserver#run()

@Overridepublic void run() {    // 1    if (outputFused) {        drainFused();    } else {        // 2        drainNormal();    }}复制代码

在注释1处会先判断outputFused这个标志位，它表示事件流是否被融化掉，默认是false，所以，最后会执行到drainNormal()方法。接着看看drainNormal()方法内部的处理。

ObserveOnObserver#drainNormal()

void drainNormal() {    int missed = 1;    final SimpleQueue
    
      q = queue;    // 1    final Observer
      a = downstream;    ...    // 2    v = q.poll();    ...    // 3    a.onNext(v);    ...}复制代码

在注释1处，这里的downstream实际上是从外面传进来的SubscribeOnObserver对象。在注释2处将队列中的消息取出来，接着在注释3处调用了SubscribeOnObserver的onNext方法。最终，会从我们包装类的最外层一直调用到最里面的我们自定义的Observer中的onNext()方法，所以，在observeOn()方法下面的链式代码都会执行到它所指定的线程中。

总结

ObservableCreate 一> ObservableSubscribeOn 一> ObservableObserveOn 为初始化顺序 当调用observable.subscribe(observer)时的执行顺序 ObservableObserveOn 一> ObservableSubscribeOn 一> ObservableCreate 当发送消息的执行顺序 ObservableCreate 一> ObservableSubscribeOn 一> ObservableObserveOn

RxJava 操作符

  • throttleFirst

Android中最常见的点击按钮，弹出一个新的Fragment的效果，其代码如下。

findViewById(R.id.normal_button).setOnClickListener(view -> {    FragmentTransaction transaction = getFragmentManager().beginTransaction();    transaction            .setCustomAnimations(R.animator.slide_up, R.animator.slide_down, R.animator.slide_up, R.animator.slide_down)            .setTransition(FragmentTransaction.TRANSIT_FRAGMENT_OPEN);    transaction            .add(android.R.id.content, new DummyFragment())            .addToBackStack(DummyFragment.class.getSimpleName()).commit();});复制代码

这段代码存在的问题在于当用户无意中多次点击到按钮时，会弹出多个Fragment。

我们通过使用throttleFirst操作符来解决按钮被多次点击的问题。throttleFirst允许设置一个时间长度，之后它会发送固定时间长度内的第一个事件，而屏蔽其它事件，在间隔达到设置的时间后，可以再发送下一个事件，如下图所示。

转载于:https://juejin.im/post/5ccee30ff265da038c02263a

你可能感兴趣的文章
Linux运维课之Mysql cluster随堂视频
查看>>
Android入门之创建一个AndroidStudio工程
查看>>
2012年下半年系统集成项目管理工程师真题（案例分析）(3)
查看>>
LAMP平台下用Drupal快速建站
查看>>
利用nginx的proxy_next_upstream实现线路容灾
查看>>
chrome 插件开发
查看>>
[LintCode] Serialize and Deserialize Binary Tree
查看>>
Android 矢量图
查看>>
linux awk命令详解
查看>>
MySQL的SET字段类型
查看>>
Quartz数据库表分析
查看>>
Linux基本命令总结
查看>>
UltraGrid中增删改功能推荐做法
查看>>
我的友情链接
查看>>
iOS 自定义 NavigationController
查看>>
Web开发中的各种色彩
查看>>
cocos2dx v3.9 与SpriteFrameCache和AnimationCache之间的对话
查看>>
说说“用户无线网络时不时断开重连”的故障！
查看>>
Vue 爬坑之路（九）—— 用正确的姿势封装组件
查看>>
我的友情链接
查看>>
喝酒易醉，品茶养心，人生如梦，品茶悟道，何以解忧？唯有杜康！-- 愿君每日到此一游！
当前时间: 2025-02-03 08:48:23   当前IP: 18.219.10.176   联系邮箱:javaeecc@qq.com     Copyright © 2020 - 2022 baihongyu.com 京ICP备2021015314号-2
强烈建议你试试无所不能的CHAT-GPT，快点击我
强烈建议你试试无所不能的CHAT-GPT，快点击我