rx-java 基本范例
示例
调度程序是有关处理单元的RxJava抽象。调度程序可以由Executor服务支持，但是您可以实现自己的调度程序实现。
AScheduler应该满足此要求：
应该顺序处理未延迟的任务（FIFO顺序）
任务可以延迟
Scheduler可以在某些运算符（例如：）中将A用作参数delay，或与subscribeOn/observeOn方法一起使用。
对于某些运算符，Scheduler将使用来处理特定运算符的任务。例如，delay将安排一个延迟任务，该任务将发出下一个值。这Scheduler将保留并在以后执行。
该subscribeOn可每使用一次Observable。它将定义Scheduler订阅的代码将在其中执行。
每个observeOn可以多次使用Observable。这将限定其中Scheduler将用于执行所定义的所有的任务之后的observeOn方法。observeOn将帮助您执行线程跳跃。
在特定的调度程序上
//该lambda将在Schedulers.io()中执行
Observable.fromCallable(() -> Thread.currentThread().getName())
.subscribeOn(Schedulers.io())
.subscribe(System.out::println);使用特定的Scheduler进行observeOn
Observable.fromCallable(() -> "Thread -> " + Thread.currentThread().getName())
//下一个任务将在io调度程序中执行
.observeOn(Schedulers.io())
.map(str -> str + " -> " + Thread.currentThread().getName())
//下一个任务将在计算调度程序中执行
.observeOn(Schedulers.computation())
.map(str -> str + " -> " + Thread.currentThread().getName())
//下一个任务将在io调度程序中执行
.observeOn(Schedulers.newThread())
.subscribe(str -> System.out.println(str + " -> " + Thread.currentThread().getName()));用运算符指定特定的调度程序
一些运算符可以将Scheduleras作为参数。
Observable.just(1)
//延迟运算符的onNext方法将在新线程中执行
.delay(1, TimeUnit.SECONDS, Schedulers.newThread())
.subscribe(System.out::println);发布给订阅者：
TestScheduler testScheduler = Schedulers.test(); EventBus sut = new DefaultEventBus(testScheduler); TestSubscriber<Event> subscriber = new TestSubscriber<Event>(); sut.get().subscribe(subscriber); sut.publish(event); testScheduler.advanceTimeBy(1, TimeUnit.SECONDS);
线程池：
this.poolName = schedulerFig.getIoSchedulerName(); final int poolSize = schedulerFig.getMaxIoThreads(); final BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(poolSize); final MaxSizeThreadPool threadPool = new MaxSizeThreadPool( queue, poolSize ); this.scheduler = Schedulers.from(threadPool);
Web套接字可观察：
final Subscription subscribe = socket.webSocketObservable()
.subscribeOn(Schedulers.io())
.doOnNext(new Action1<RxEvent>() {
@Override
public void call(RxEvent rxEvent) {
System.out.println("Event: " + rxEvent);
}
})
.subscribe();