RxJava 教程第四部分：并发 之线程调度

final BehaviorSubject<Integer> subject = BehaviorSubject.create(); subject.subscribe(i -> {     System.out.println("Received " + i + " on " + Thread.currentThread().getId()); });   int[] i = {1}; // naughty side-effects for examples only ;) Runnable r = () -> {     synchronized(i) {         System.out.println("onNext(" + i[0] + ") on " + Thread.currentThread().getId());         subject.onNext(i[0]++);     } };   r.run(); // Execute on main thread new Thread(r).start(); new Thread(r).start();   

onNext(1) on 1 Received 1 on 1 onNext(2) on 11 Received 2 on 11 onNext(3) on 12 Received 3 on 12   

public final Observable<T> observeOn(Schedulerscheduler) public final Observable<T> subscribeOn(Schedulerscheduler)   

System.out.println("Main: " + Thread.currentThread().getId());   Observable.create(o -> {         System.out.println("Created on " + Thread.currentThread().getId());         o.onNext(1);         o.onNext(2);         o.onCompleted();     })     //.subscribeOn(Schedulers.newThread())     .subscribe(i -> {         System.out.println("Received " + i + " on " + Thread.currentThread().getId());     });   System.out.println("Finished main: " + Thread.currentThread().getId());   

Main: 1 Createdon 1 Received 1 on 1 Received 2 on 1 Finishedmain: 1   

Main: 1 Finishedmain: 1 Createdon 11 Received 1 on 11 Received 2 on 11   

System.out.println("Main: " + Thread.currentThread().getId());   Observable.interval(100, TimeUnit.MILLISECONDS)     .subscribe(i -> {         System.out.println("Received " + i + " on " + Thread.currentThread().getId());     });   System.out.println("Finished main: " + Thread.currentThread().getId());   

Main: 1 Finishedmain: 1 Received 0 on 11 Received 1 on 11 Received 2 on 11   

Observable.create(o -> {         System.out.println("Created on " + Thread.currentThread().getId());         o.onNext(1);         o.onNext(2);         o.onCompleted();     })     .observeOn(Schedulers.newThread())     .subscribe(i ->         System.out.println("Received " + i + " on " + Thread.currentThread().getId()));   

Createdon 1 Received 1 on 13 Received 2 on 13   

Observable.create(o -> {         System.out.println("Created on " + Thread.currentThread().getId());         o.onNext(1);         o.onNext(2);         o.onCompleted();     })     .doOnNext(i ->          System.out.println("Before " + i + " on " + Thread.currentThread().getId()))     .observeOn(Schedulers.newThread())     .doOnNext(i ->          System.out.println("After " + i + " on " + Thread.currentThread().getId()))     .subscribe();   

Createdon 1 Before 1 on 1 Before 2 on 1 After 1 on 13 After 2 on 13   

Observable<Object> source = Observable.using(     () -> {         System.out.println("Subscribed on " + Thread.currentThread().getId());         return Arrays.asList(1,2);     },     (ints) -> {         System.out.println("Producing on " + Thread.currentThread().getId());         return Observable.from(ints);     },     (ints) -> {         System.out.println("Unubscribed on " + Thread.currentThread().getId());     } );   source     .unsubscribeOn(Schedulers.newThread())     .subscribe(System.out::println);   

Subscribedon 1 Producingon 1 1 2 Unubscribedon 11   

RxJava 内置的 Scheduler 有：

– immediate 同步执行

– trampoline 把任务放到当前线程的队列中，等当前任务执行完了，再继续执行队列中的任务

– newThread 对于每个任务创建一个新的线程去执行

– computation 计算线程，用于需要大量 CPU 计算的任务

– io 用于执行 io 操作的任务

– test 用于测试和调试

Scheduler.Workerworker = scheduler.createWorker(); worker.schedule(     () -> System.out.println("Action"));   

Subscriptionschedule(     Action0action,     long delayTime,     java.util.concurrent.TimeUnitunit) SubscriptionschedulePeriodically(     Action0action,     long initialDelay,     long period,     java.util.concurrent.TimeUnitunit)   

Schedulerscheduler = Schedulers.newThread(); long start = System.currentTimeMillis(); Scheduler.Workerworker = scheduler.createWorker(); worker.schedule(     () -> System.out.println(System.currentTimeMillis()-start),     5, TimeUnit.SECONDS); worker.schedule(     () -> System.out.println(System.currentTimeMillis()-start),     5, TimeUnit.SECONDS);   

5033 5035   

Schedulerscheduler = Schedulers.newThread(); long start = System.currentTimeMillis(); Scheduler.Workerworker = scheduler.createWorker(); worker.schedule(     () -> {         System.out.println(System.currentTimeMillis()-start);         worker.unsubscribe();     },     5, TimeUnit.SECONDS); worker.schedule(     () -> System.out.println(System.currentTimeMillis()-start),     5, TimeUnit.SECONDS);   

5032   

Schedulerscheduler = Schedulers.newThread(); long start = System.currentTimeMillis(); Scheduler.Workerworker = scheduler.createWorker(); worker.schedule(() -> {     try {         Thread.sleep(2000);         System.out.println("Action completed");     } catch (InterruptedException e) {         System.out.println("Action interrupted");     } }); Thread.sleep(500); worker.unsubscribe();   

Actioninterrupted   

Schedulerscheduler = Schedulers.immediate(); Scheduler.Workerworker = scheduler.createWorker(); worker.schedule(() -> {     System.out.println("Start");     worker.schedule(() -> System.out.println("Inner"));     System.out.println("End"); });   

Start Inner End   

Schedulerscheduler = Schedulers.trampoline(); Scheduler.Workerworker = scheduler.createWorker(); worker.schedule(() -> {     System.out.println("Start");     worker.schedule(() -> System.out.println("Inner"));     System.out.println("End"); });   

Start End Inner   

public static void printThread(String message) {     System.out.println(message + " on " + Thread.currentThread().getId()); }   

printThread("Main"); Schedulerscheduler = Schedulers.newThread(); Scheduler.Workerworker = scheduler.createWorker(); worker.schedule(() -> {     printThread("Start");     worker.schedule(() -> printThread("Inner"));     printThread("End"); }); Thread.sleep(500); worker.schedule(() -> printThread("Again"));   

Mainon 1 Starton 11 End on 11 Inneron 11 Againon 11