本篇文章主要是阅读官网的scheduler文档。

弹珠图

弹珠图可以很好的描述rxjava的运作流程，建议先看下这篇文章Understanding Marble Diagrams for Reactive Streams

简单scheduler

输出结果: RxNewThreadScheduler-1 hello world!

  1. public class SchedulersWork {
  2.     private static final CountDownLatch latch = new CountDownLatch(1);
  4.     public static void main(String[] args) throws InterruptedException {
  5.         Scheduler.Worker worker = Schedulers.newThread().createWorker();
  6.         worker.schedule(SchedulersWork::sayHello);
  7.         latch.await();
  8.         worker.dispose();
  9.     }
  11.     public static void sayHello() {
  12.         System.out.println(Thread.currentThread().getName() + " hello world!");
  13.         latch.countDown();
  14.     }
  15. }

递归scheduler

输出结果: RxNewThreadScheduler-1 hello world! …
循环输出直到dispose worker

  1. public class RecursiveScheduler {
  2.     public static void main(String[] args) throws InterruptedException {
  3.         Scheduler.Worker worker = Schedulers.newThread().createWorker();
  4.         worker.schedule(new Runnable() {
  5.             @Override
  6.             public void run() {
  7.                 sayHello();
  8.                 //递归直到dispose
  9.                 worker.schedule(this);
  10.             }
  11.         });
  13.         Thread.sleep(1000);
  14.         worker.dispose();
  15.     }
  17.     public static void sayHello() {
  18.         System.out.println(Thread.currentThread().getName() + " hello world!");
  19.     }
  20. }

⚠️: 递归调用需要限制递归次数或者主动设置dispose状态，否则会出现死循环。

检查或者设置dispose状态

  1. package com.zihao.schedulers;
  3. import io.reactivex.rxjava3.core.Scheduler;
  4. import io.reactivex.rxjava3.schedulers.Schedulers;
  6. import java.util.Scanner;
  8. /**
  9.  * 相比较普通的recursive 调度器，使用dispose状态检查
  10.  * 如果不订阅了 就停止并释放资源
  11.  *
  12.  * @author tangzihao
  13.  * @Date 2021/1/3 11:20 上午
  14.  */
  15. public class CheckDisposeRecursiveScheduler {
  16.     public static void main(String[] args) throws InterruptedException {
  17.         Scheduler.Worker worker = Schedulers.newThread().createWorker();
  18.         worker.schedule(new Runnable() {
  19.             @Override
  20.             public void run() {
  21.                 while (!worker.isDisposed()) {
  22.                     sayHello();
  23.                 }
  24.                 System.out.println("worker被终止了。。。");
  25.             }
  26.         });
  28.         //主线程通过终端控制worker结束
  29.         Scanner scan = new Scanner(System.in);
  30.         System.out.println("终止worker工作: ");
  31.         if (scan.hasNext()) {
  32.             String str = scan.next();
  33.             if (str.equals("stop")) {
  34.                 worker.dispose();
  35.             }
  36.         }
  37.         scan.close();
  38.     }
  40.     public static void sayHello() {
  41.         //do nothing 故意不输出方便终端输入
  42.     }
  43. }

输出结果

  1. 终止worker工作: 
  2. stop  //在控制台输入 main线程
  3. worker被终止了。。。 //worker线程

延迟或周期性调度器

delayed scheduler

schedule有三个参数，调用的方法，延迟的时间，时间单位

  1. public class DelayedAndPeriodicScheduler {
  2.     public static void main(String[] args) throws InterruptedException {
  3.         Scheduler.Worker worker = Schedulers.newThread().createWorker();
  4.         worker.schedule(DelayedAndPeriodicScheduler::sayHello,1, TimeUnit.SECONDS);
  5.         Thread.sleep(2000);
  6.     }
  8.     public static void sayHello(){
  9.         System.out.println("hello,world!");
  10.     }
  11. }

periodic scheduler

schedulePeriodically有四个参数，调用的方法，延迟的时间，周期性时间，时间单位

  1. public class DelayedAndPeriodicScheduler {
  2.     public static void main(String[] args) throws InterruptedException {
  3.         periodicScheduler();
  4.     }
  6.     public static void periodicScheduler() throws InterruptedException {
  7.         Scheduler.Worker worker = Schedulers.newThread().createWorker();
  8.         worker.schedulePeriodically(DelayedAndPeriodicScheduler::sayHello, 500, 250, TimeUnit.MILLISECONDS);
  9.         Thread.sleep(3000);
  10.     }
  12.     public static void sayHello() {
  13.         System.out.println("hello,world!");
  14.     }
  15. }