Scheduler

优质
小牛编辑
128浏览
2023-12-01

如果你想给Observable操作符链添加多线程功能,你可以指定操作符(或者特定的Observable)在特定的调度器(Scheduler)上执行。

某些ReactiveX的Observable操作符有一些变体,它们可以接受一个Scheduler参数。这个参数指定操作符将它们的部分或全部任务放在一个特定的调度器上执行。

使用ObserveOn和SubscribeOn操作符,你可以让Observable在一个特定的调度器上执行,ObserveOn指示一个Observable在一个特定的调度器上调用观察者的onNext, onError和onCompleted方法,SubscribeOn更进一步,它指示Observable将全部的处理过程(包括发射数据和通知)放在特定的调度器上执行。

RxJava示例

调度器的种类

下表展示了RxJava中可用的调度器种类:

调度器类型效果
Schedulers.computation( )用于计算任务,如事件循环或和回调处理,不要用于IO操作(IO操作请使用Schedulers.io());默认线程数等于处理器的数量
Schedulers.from(executor)使用指定的Executor作为调度器
Schedulers.immediate( )在当前线程立即开始执行任务
Schedulers.io( )用于IO密集型任务,如异步阻塞IO操作,这个调度器的线程池会根据需要增长;对于普通的计算任务,请使用Schedulers.computation();Schedulers.io( )默认是一个CachedThreadScheduler,很像一个有线程缓存的新线程调度器
Schedulers.newThread( )为每个任务创建一个新线程
Schedulers.trampoline( )当其它排队的任务完成后,在当前线程排队开始执行

默认调度器

在RxJava中,某些Observable操作符的变体允许你设置用于操作执行的调度器,其它的则不在任何特定的调度器上执行,或者在一个指定的默认调度器上执行。下面的表格个列出了一些操作符的默认调度器:

操作符调度器
buffer(timespan)computation
buffer(timespan, count)computation
buffer(timespan, timeshift)computation
debounce(timeout, unit)computation
delay(delay, unit)computation
delaySubscription(delay, unit)computation
intervalcomputation
repeattrampoline
replay(time, unit)computation
replay(buffersize, time, unit)computation
replay(selector, time, unit)computation
replay(selector, buffersize, time, unit)computation
retrytrampoline
sample(period, unit)computation
skip(time, unit)computation
skipLast(time, unit)computation
take(time, unit)computation
takeLast(time, unit)computation
takeLast(count, time, unit)computation
takeLastBuffer(time, unit)computation
takeLastBuffer(count, time, unit)computation
throttleFirstcomputation
throttleLastcomputation
throttleWithTimeoutcomputation
timeIntervalimmediate
timeout(timeoutSelector)immediate
timeout(firstTimeoutSelector, timeoutSelector)immediate
timeout(timeoutSelector, other)immediate
timeout(timeout, timeUnit)computation
timeout(firstTimeoutSelector, timeoutSelector, other)immediate
timeout(timeout, timeUnit, other)computation
timercomputation
timestampimmediate
window(timespan)computation
window(timespan, count)computation
window(timespan, timeshift)computation

使用调度器

除了将这些调度器传递给RxJava的Observable操作符,你也可以用它们调度你自己的任务。下面的示例展示了Scheduler.Worker的用法:

  1. worker = Schedulers.newThread().createWorker();
  2. worker.schedule(new Action0() {
  3. @Override
  4. public void call() {
  5. yourWork();
  6. }
  7. });
  8. // some time later...
  9. worker.unsubscribe();

递归调度器

要调度递归的方法调用,你可以使用schedule,然后再用schedule(this),示例:

  1. worker = Schedulers.newThread().createWorker();
  2. worker.schedule(new Action0() {
  3. @Override
  4. public void call() {
  5. yourWork();
  6. // recurse until unsubscribed (schedule will do nothing if unsubscribed)
  7. worker.schedule(this);
  8. }
  9. });
  10. // some time later...
  11. worker.unsubscribe();

检查或设置取消订阅状态

Worker类的对象实现了Subscription接口,使用它的isUnsubscribed和unsubscribe方法,所以你可以在订阅取消时停止任务,或者从正在调度的任务内部取消订阅,示例:

  1. Worker worker = Schedulers.newThread().createWorker();
  2. Subscription mySubscription = worker.schedule(new Action0() {
  3. @Override
  4. public void call() {
  5. while(!worker.isUnsubscribed()) {
  6. status = yourWork();
  7. if(QUIT == status) { worker.unsubscribe(); }
  8. }
  9. }
  10. });

Worker同时是Subscription,因此你可以(通常也应该)调用它的unsubscribe方法通知可以挂起任务和释放资源了。

延时和周期调度器

你可以使用schedule(action,delayTime,timeUnit)在指定的调度器上延时执行你的任务,下面例子中的任务将在500毫秒之后开始执行:

  1. someScheduler.schedule(someAction, 500, TimeUnit.MILLISECONDS);

使用另一个版本的schedule,schedulePeriodically(action,initialDelay,period,timeUnit)方法让你可以安排一个定期执行的任务,下面例子的任务将在500毫秒之后执行,然后每250毫秒执行一次:

  1. someScheduler.schedulePeriodically(someAction, 500, 250, TimeUnit.MILLISECONDS);

测试调度器

TestScheduler让你可以对调度器的时钟表现进行手动微调。这对依赖精确时间安排的任务的测试很有用处。这个调度器有三个额外的方法:

  • advanceTimeTo(time,unit) 向前波动调度器的时钟到一个指定的时间点
  • advanceTimeBy(time,unit) 将调度器的时钟向前拨动一个指定的时间段
  • triggerActions( ) 开始执行任何计划中的但是未启动的任务,如果它们的计划时间等于或者早于调度器时钟的当前时间

最后更新:

类似资料

  • http://reactivex.io/intro.html Rx介绍 ReactiveX的历史 ReactiveX是Reactive Extensions的缩写,一般简写为Rx,最初是LINQ的一个扩展,由微软的架构师Erik Meijer领导的团队开发,在2012年11月开源,Rx是一个编程模型,目标是提供一致的编程接口,帮助开发者更方便的处理异步数据流,Rx库支持.NET、JavaScrip

  • hyperf/reactive-x 组件提供了 Swoole/Hyperf 环境下的 ReactiveX 集成。 ReactiveX 的历史 ReactiveX 是 Reactive Extensions 的缩写,一般简写为 Rx,最初是 LINQ 的一个扩展,由微软的架构师 Erik Meijer 领导的团队开发,在 2012 年 11 月开源,Rx 是一个编程模型,目标是提供一致的编程接口,帮

  • RxPY 是 Reactive Extensions 的 Python 版本。

  • ReactiveX 是 Reactive Extensions 的缩写,一般简写为 Rx,最初是 LINQ 的一个扩展,由微软的架构师 Erik Meijer 领导的团队开发,在2012年11月开源,Rx 是一个编程模型,目标是提供一致的编程接口,帮助开发者更方便的处理异步数据流,Rx库支持.NET、JavaScript和C++,Rx近几年越来越流行了,现在已经支持几乎全部的流行编程语言了,Rx的

  • 我试图使用testNG运行suite,但我遇到了一个错误:我在这个网站和web上搜索了一些关于这个问题的信息,但没有成功。 测试: 我在网上寻找答案,但没有成功。 有人能帮我解决这个问题吗?

  • 问题内容: 我是ReactiveX for Java的新手,我有以下代码块可以进行外部http调用,但它不是异步的。我们正在使用rxjava 1.2和Java 1.8 我有以下在网上找到的代码块,但我无法完全理解它,以及如何将其应用于代码库。 问题答案: 如果我对您的理解正确,则需要使用类似的方法包装现有的内容 代码的简短说明: 它计划在 对成功案例和错误案例的转换最少。它也发生在调度程序上,但是

开发工具

ReactiveX RxPY