文章目录 Rxjava && Rxandroid 概念 流程图 代码示例 Observable Observable#subscribeOn(@NonNull Scheduler scheduler) Observable#observeOn(@NonNull Scheduler scheduler) ObservableSubscribeOn ObservableObserveOn 本文开头的代码示例等同于如下代码 Schedulers调度器 internal.schedulers.ScheduledRunnable IoScheduler IoScheduler.ThreadWorker IoScheduler.CachedWorkerPool IoScheduler.EventLoopWorker ComputationScheduler AndroidSchedulers 和 HandlerScheduler Flowable & Backpressure背压 Operators 操作符 创建操作符类型 变换操作 过滤操作 布尔变量操作 算术操作 To系列 其他 参考资料
implementation "io.reactivex.rxjava3:rxjava:3.1.1"
implementation 'io.reactivex.rxjava3:rxandroid:3.0.0'
Observable是被观察者，也是生产者，也是source类型。Observer是观察者，也是消费者。被观察者、生产者、source、上游、upstream 是一类概念。在subscribeOn和observeOn经常会看到source成员变量。一般是指的ObservableSource<T>接口类型，ObservableSource实现了该接口。 观察者、消费者、下游、downstream 是一类概念。
代码功能：被观察者（生产者、源端）使用 Schedulers.io()线程池执行，观察者（消费者）用Android主线程AndroidSchedulers.mainThread()执行。处理String类型。 在步骤 0 Observable#subscribe(Observer<? super T>) 函数执行之后，会引发后续的步骤 1, 2, 3, 4, 5执行。 Observable . create ( new ObservableOnSubscribe < String > ( ) { @Override public void subscribe ( @NonNull ObservableEmitter < String > ) throws Throwable { emitter. onNext ( "emitter string 1" ) ; emitter. onNext ( "emitter string 2" ) ; emitter. onNext ( "emitter string 3" ) ; emitter. onComplete ( ) ; }
} ) . subscribeOn ( Schedulers . io ( ) ) . observeOn ( AndroidSchedulers . mainThread ( ) ) . subscribe ( new Observer < String > ( ) { @Override public void onSubscribe ( @NonNull Disposable d) { Log . i ( TAG, "onSubscribe " + d. getClass ( ) ) ; } @Override public void onNext ( @NonNull String s) { Log . i ( TAG, "onNext " + s) ; } @Override public void onError ( @NonNull Throwable e) { Log . i ( TAG, "onError " + e) ; e. printStackTrace ( ) ; } @Override public void onComplete ( ) { Log . i ( TAG, "onComplete" ) ; } } ) ;
关键入口类。 关键入口方法: Observable#create(@NonNull ObservableOnSubscribe<T> source),生成一个Observable<T>对象。实际类型是ObservableCreate<T>类型。 唯一实现的接口是：ObservableSource。 subscribeOn 决定被观察者的线程环境。Task任务发生、执行的线程环境。subscribeOn如果连续两次调用该函数，仅仅第一次调用配置生效。 因为被观察者(生产者)是从下游向上游触发，代码的第一次配置subscribeOn，实际对应触发流程的最后一个步骤，因此也只有第一次配置生效。每次调用，该函数都会生成一个Observable<T>对象。实际类型是ObservableSubscribeOn<T>。ObservableSubscribeOn<T>持有source成员变量，source的类型是 ObservableSource<T>，实际类型是Observable。 observeOn 决定观察者的线程环境。通过哪个线程通知到观察者Observer。observeOn在调用链中的每一次调用，都会导致后续的observer操作切换线程。观察者(消费者)是由被观察者(生产者)从上游向下游触发，每一次observeOn都会导致一次线程切换。每次调用，该函数都会生成一个Observable<T>对象。实际类型是ObservableObserveOn<T>。ObservableObserveOn<T>持有source成员变量，source的类型是 ObservableSource<T>，实际类型是Observable。 含有一个内部Observer类型，内部类：ObservableSubscribeOn.SubscribeOnObserver<T>。用于注册subscribe上游的观察者对象 第一处注释实现了生产者执行线程的切换。 第二处注释，通过subscribe函数，注册上游观察者对象。 源码: ObservableSubscribeOn #subscribeActual
@Override
public void subscribeActual ( final Observer < ? super T > ) { final SubscribeOnObserver < T > = new SubscribeOnObserver < > ( observer) ; observer. onSubscribe ( parent) ; parent. setDisposable ( scheduler. scheduleDirect ( new SubscribeTask ( parent) ) ) ;
}
源码: ObservableSubscribeOn. SubscribeTask
final class SubscribeTask implements Runnable { private final SubscribeOnObserver < T > ; SubscribeTask ( SubscribeOnObserver < T > ) { this . parent = parent; } @Override public void run ( ) { source. subscribe ( parent) ; }
}
含有一个内部Observer类型，内部类：ObservableObserveOn.ObserveOnObserver<T>。 第一处注释，生成了消费者的子线程worker 第二处注释，通过subscribe函数，注册上游观察者对象。 源码: ObservableObserveOn #subscribeActual
@Override
protected void subscribeActual ( Observer < ? super T > ) { if ( scheduler instanceof TrampolineScheduler ) { source. subscribe ( observer) ; } else { Scheduler. Worker w = scheduler. createWorker ( ) ; source. subscribe ( new ObserveOnObserver < > ( observer, w, delayError, bufferSize) ) ; }
}
ObservableCreate < String > = ( ObservableCreate < String > ) Observable . create ( new ObservableOnSubscribe < String > ( ) { @Override public void subscribe ( @NonNull ObservableEmitter < String > ) throws Throwable { emitter. onNext ( "emitter string 1" ) ; emitter. onNext ( "emitter string 2" ) ; emitter. onNext ( "emitter string 3" ) ; emitter. onComplete ( ) ; }
} ) ;
ObservableSubscribeOn < String > = ( ObservableSubscribeOn < String > ) observableCreate. subscribeOn ( Schedulers . io ( ) ) ;
ObservableObserveOn < String > = ( ObservableObserveOn < String > ) subscribeOn. observeOn ( AndroidSchedulers . mainThread ( ) ) ;
observeOn. subscribe ( new Observer < String > ( ) { @Override public void onSubscribe ( @NonNull Disposable d) { Log . i ( TAG, "onSubscribe " + d. getClass ( ) ) ; } @Override public void onComplete ( ) { Log . i ( TAG, "onComplete" ) ; }
} ) ;
Scheduler . io ( )
Scheduler . computation ( )
Scheduler . trampoline ( )
Scheduler . newThread ( )
Scheduler . single ( )
Scheduler . from ( executor)
AndroidSchedulers . mainThread ( )
internal.schedulers.ScheduledRunnable
实现了java.util.concurrent.Callable接口，用于线程池执行。实现了disposables.Disposable接口，用于执行中取消Futrue。 默认的IO调度器。默认使用的是调度线程池ScheduledThreadPoolExecutor。参考函数: SchedulerPoolFactory#create Runnable对象最终是在ThreadWorker中对应的线程池中执行的。 CachedWorkerPool作为对象池，缓存ThreadWorker。单线程的，定时的，顺序执行任务的worker。一个worker可以执行很多的task。 继承自 NewThreadWorker。每一个ThreadWorker都对应一个ScheduledThreadPool调度线程池。 ScheduledThreadPool该线程池的核心线程数为1(且不会增长)，所有事件都会被抛到DelayedWorkQueue等待调度执行。ThreadWorker的对象池。 allWorkers正在运行的worker。allWorkers是CompositeDisposable类型。CompositeDisposable单词的意思是Disposable容器，这个类是一个用OpenHashSet存储的容器。IoScheduler.CachedWorkerPool#get()方法会线程尝试复用expiringWorkerQueue队列中的对象。否则，创建一个新对象。expiringWorkerQueue过期的ThreadWorker对象，可以复用。通过IoScheduler.CachedWorkerPool#release函数回收到expiringWorkerQueue队列中。被回收到expiringWorkerQueue 60秒之后，ThreadWorker会被彻底过期，而被移除。移除函数是CachedWorkerPool#evictExpiredWorkers。evictExpiredWorkers：allWorkers是CompositeDisposable类型，remove函数内部调用dispose停止线程池。static void evictExpiredWorkers ( ConcurrentLinkedQueue < ThreadWorker > , CompositeDisposable allWorkers) { if ( ! expiringWorkerQueue. isEmpty ( ) ) { long currentTimestamp = now ( ) ; for ( ThreadWorker threadWorker : expiringWorkerQueue) { if ( threadWorker. getExpirationTime ( ) <= currentTimestamp) { if ( expiringWorkerQueue. remove ( threadWorker) ) { allWorkers. remove ( threadWorker) ; } } else { break ; } } }
}
IoScheduler.EventLoopWorker封装了一层 ThreadWorker。IoScheduler#createWorker创建的是该对象。此处是类似于代理模式或者装饰器模式。EventLoopWorker设计的目的是调用disposable函数销毁时，不同时销毁内部的ThreadWorker对象。经由EventLoopWorker执行的Runnable任务都会被存储到IoScheduler.EventLoopWorker#tasks容器中。EventLoopWorker销毁时，经由它创建的任务也一并销毁。 强烈建议参考对PoolWorker和EventLoopWorker的解释。https://juejin.cn/post/6844903657792634894 PoolWorker继承自 NewThreadWorker。每个PoolWorker对应一个ScheduledThreadPool调度线程池。ComputationScheduler.FixedSchedulerPool是一个scheduler对象池，包含有固定数量（内核CPU数）的PoolWorker。成员变量FixedSchedulerPool#cores代表了当前线程池的大小。io. reactivex. rxjava3. internal. schedulers. ComputationScheduler. FixedSchedulerPoolFixedSchedulerPool
final int cores;
final PoolWorker [ ] eventLoops;
long n;
FixedSchedulerPool ( int maxThreads, ThreadFactory threadFactory) { this . cores = maxThreads; this . eventLoops = new PoolWorker [ maxThreads] ; for ( int i = 0 ; i < maxThreads; i++ ) { this . eventLoops[ i] = new PoolWorker ( threadFactory) ; }
}
在生成PoolWorker时，会直接从eventLoops数组获取，从0依次累加。计数变量是n。数组下标通过取余数获得。 根据上面的取余数的设计，一个PoolWorker可能会对应多个EventLoopWorker对象。代理模式，每个EventLoopWorker通过disposable销毁时，只销毁自己创建的任务，不影响其他的EventLoopWorker。 io. reactivex. rxjava3. internal. schedulers. ComputationScheduler. FixedSchedulerPoolpublic PoolWorker getEventLoop ( ) { int c = cores; if ( c == 0 ) { return SHUTDOWN_WORKER; } return eventLoops[ ( int ) ( n++ % c) ] ; }
实际就是一个采用MainLooper的Handler类，向主线程发送Runnable任务执行。 可以通过AndroidSchedulers#from(Looper looper, boolean async) 定义自己的Looper线程。 当生产者事件的发送速度远大于消费者处理的速度时，事件就会在线程池的事件调度队列中积压。当积压足够多时，会造成OOM。描述这种情况，称之为Backpressure。 BackpressureStrategy是一种控制生产者事件发送速度的策略。默认的队列大小是128。几种策略可供选择MISSING、ERROR、BUFFER、DROP、LATEST。org.reactivestreams.Subscription#request(long n)请求N个数据。参考文章,讲述的比较清晰 https://juejin.cn/post/6844903454067032071 通过request请求控制发送事件数量的代码示例。 Flowable . just ( 1 , 2 , 3 , 4 ) . subscribeOn ( Schedulers . io ( ) ) . observeOn ( AndroidSchedulers . mainThread ( ) ) . subscribe ( new Subscriber < Integer > ( ) { @Override public void onSubscribe ( Subscription s) { s. request ( 2 ) ; } @Override public void onNext ( Integer integer) { Log . i ( TAG, "onNext " + integer) ; } @Override public void onError ( Throwable t) { } @Override public void onComplete ( ) { Log . i ( TAG, "onComplete " ) ; } } ) ;
操作符分为几类，常见的有创建、变换、过滤、组合、算术和聚合、转换操作等等。 create常用的创建类型操作符。例如：Observable#create(ObservableOnSubscribe<T> source),Flowable#create(FlowableOnSubscribe<T> source, BackpressureStrategy mode)from支持将 Iterable、数组、Future 转换为Observable。例如：fromArray、fromFuture、fromIterable等just支持将一些元素，依次发送。interval按照固定的时间间隔发送数据。interval默认在computation调度器上执行。map ( @NonNull Function < ? super T , ? extends R > )
flatMap一个 Obervable 转换成多个 Observable进行发射。flat的含义是水平的，平铺的含义。flatMap输出是无序的。内部采用merge动作合并public final < @NonNull R > Observable < R > flatMap ( @NonNull Function < ? super T , ? extends ObservableSource < ? extends R > > )
concatMap操作符的功能和flatMap是非常相似的，区别是：concatMap 输出和原始数据序列一致。concatMap输出是有序的。内部采用concat动作合并public final < @NonNull R > Observable < R > concatMap ( @NonNull Function < ? super T , ? extends ObservableSource < ? extends R > > )
filter过滤符合条件的类型。ofType过滤指定的class类型。ofType(@NonNull Class<U> clazz)distinct去重操作。通过HashSet实现的去重功能take(long count)和takeLast(long count)。take是获取前count个数据。 takeLast是获取后count个数据all所有的数据都符合某个条件，返回True。contains包含某个数据，返回True。isEmpty没有数据要发送，返回True。Average、ConcatMax、Min、Count、SumObservable#toFlowable(@NonNull BackpressureStrategy strategy)转换为Flowable。Flowable#toObservable()转换为Observable。toMap、toList、toIterable转换为map、list、Iterable。plugins.RxJavaPlugins 插件类，用于自定义的一些接口。默认情况下，没有任何定制插件Function对象，会直接返回入参。Disposable 表示可支持取消的，一次性的执行任务。扩展出了很多的子类实现。在和线程池使用是，主要是包装了Futrue<T>类型，取消动作通过Futrue<T>实现。 java.util.concurrent.ScheduledThreadPoolExecutor 任务定时执行线程池 java.util.concurrent.ThreadFactory 线程Fractory接口 java.util.concurrent.atomic.AtomicReference 原子化对象类 java.util.concurrent.atomic.AtomicReferenceArray 原子化数组类 https://www.jianshu.com/p/a3f2c3ee00a3 
