public final class ComputationScheduler extends Scheduler implements SchedulerMultiWorkerSupport {
  // 资源池
  final AtomicReference<FixedSchedulerPool> pool;
  // 这是computationScheduler类中实现的createWork()方法
  public Worker createWorker() {
    // 创建EventLoop工作者，入参是一个PoolWorker
      return new EventLoopWorker(pool.get().getEventLoop());
  }
static final class FixedSchedulerPool implements SchedulerMultiWorkerSupport {
        final int cores;
        // 资源池工作者，每个工作者其实都是一个定时线程池
        final PoolWorker[] eventLoops;
        long n;
        // 对应前面的函数调用
        public PoolWorker getEventLoop() {
          int c = cores;
          if (c == 0) {
              return SHUTDOWN_WORKER;
          }
          // simple round robin, improvements to come
          // 这里其实就是从工作者数组中轮询选出一个工作者
          这里其实拥有提升和优化的空间，这里笔者可能会向开源社区提交一个pr
          以此进行比较好的调度器调度
          return eventLoops[(int)(n++ % c)];
        }
// 此处是一个简单的封装        
static final class PoolWorker extends NewThreadWorker {
        PoolWorker(ThreadFactory threadFactory) {
            super(threadFactory);
        }
    }
public class NewThreadWorker extends Scheduler.Worker implements Disposable {
  private final ScheduledExecutorService executor;
  volatile boolean disposed;
  public NewThreadWorker(ThreadFactory threadFactory) {
      // 进行定时线程池的初始化
      executor = SchedulerPoolFactory.create(threadFactory);
  }
  public static ScheduledExecutorService create(ThreadFactory factory) {
    final ScheduledExecutorService exec =
    // 初始化一个定时线程池
    Executors.newScheduledThreadPool(1, factory);
    tryPutIntoPool(PURGE_ENABLED, exec);
    return exec;
  }

// 还是从subscribeActul开始（原因见上一篇文章）
public void subscribeActual(Subscriber<? super T> s) {
    Worker worker = scheduler.createWorker();
    if (s instanceof ConditionalSubscriber) {
        source.subscribe(new ObserveOnConditionalSubscriber<T>(
                (ConditionalSubscriber<? super T>) s, worker, delayError, prefetch));
    } else {
        // 其内部封装了一个ObserveOnsubcriber，这是个对下流订阅者的封装，主要什么作用呢，为什么要这个呢，
        // 其实这个涉及订阅线程内部的机制，接着看源代码了解其内部机制
        source.subscribe(new ObserveOnSubscriber<T>(s, worker, delayError, prefetch));
    }
}
  // 基类
  abstract static class BaseObserveOnSubscriber<T>
  extends BasicIntQueueSubscription<T>
  implements FlowableSubscriber<T>, Runnable {
      private static final long serialVersionUID = -8241002408341274697L;
      final Worker worker;
      final boolean delayError;
      final int prefetch;
      final int limit;
      final AtomicLong requested;
      Subscription upstream;
      SimpleQueue<T> queue;
      volatile boolean cancelled;
      volatile boolean done;
      Throwable error;
      int sourceMode;
      long produced;
      boolean outputFused;
      BaseObserveOnSubscriber(
              Worker worker,
              boolean delayError,
              int prefetch) {
          this.worker = worker;
          this.delayError = delayError;
          this.prefetch = prefetch;
          this.requested = new AtomicLong();
          this.limit = prefetch - (prefetch >> 2);
      }
      @Override
      public final void onNext(T t) {
          if (done) {
              return;
          }
          if (sourceMode == ASYNC) {
              trySchedule();
              return;
          }
          /**
          * 当上游的装修者（上一篇提到的装修者模式）调用onNext方 * 法时，这时并没有类似的去调用下游的onNext方法，那这个
          * 时候其实就是订阅者线程模式的核心原理：采用queue队列
          * 进行数据的store，这里尝试将数据放进队列
          */
          if (!queue.offer(t)) {
              upstream.cancel();
              error = new MissingBackpressureException("Queue is full?!");
              done = true;
          }
          // 开启订阅者线程池模式的调度，具体实现在子类中实现
          trySchedule();
      }
      @Override
      public final void onError(Throwable t) {
          if (done) {
              RxJavaPlugins.onError(t);
              return;
          }
          error = t;
          done = true;
          trySchedule();
      }
      @Override
      public final void onComplete() {
          if (!done) {
              done = true;
              trySchedule();
          }
      }
      // 这里并没有向上传递request请求，而是把自己当做数据发射者进行request计数
      @Override
      public final void request(long n) {
          if (SubscriptionHelper.validate(n)) {
              BackpressureHelper.add(requested, n);
              // 开启调度
              trySchedule();
          }
      }
      @Override
      public final void cancel() {
          if (cancelled) {
              return;
          }
          cancelled = true;
          upstream.cancel();
          worker.dispose();
          if (getAndIncrement() == 0) {
              queue.clear();
          }
      }
      // 调度代码
      final void trySchedule() {
          // 上一篇文章讲过这个的用法
          if (getAndIncrement() != 0) {
              return;
          }
          // 启用一个work来进行任务的执行 this对象说明实现了runable接口
          worker.schedule(this);
      }
      // 调度实现的代码
      @Override
      public final void run() {
          if (outputFused) {
              runBackfused();
          } else if (sourceMode == SYNC) {
              runSync();
          } else {
              // 一般会调用runAsync方法
              runAsync();
          }
      }
      abstract void runBackfused();
      abstract void runSync();
      abstract void runAsync();
      final boolean checkTerminated(boolean d, boolean empty, Subscriber<?> a) {
          if (cancelled) {
              clear();
              return true;
          }
          if (d) {
              if (delayError) {
                  if (empty) {
                      cancelled = true;
                      Throwable e = error;
                      if (e != null) {
                          a.onError(e);
                      } else {
                          a.onComplete();
                      }
                      worker.dispose();
                      return true;
                  }
              } else {
                  Throwable e = error;
                  if (e != null) {
                      cancelled = true;
                      clear();
                      a.onError(e);
                      worker.dispose();
                      return true;
                  } else
                  if (empty) {
                      cancelled = true;
                      a.onComplete();
                      worker.dispose();
                      return true;
                  }
              }
          }
          return false;
      }
      @Override
      public final int requestFusion(int requestedMode) {
          if ((requestedMode & ASYNC) != 0) {
              outputFused = true;
              return ASYNC;
          }
          return NONE;
      }
      @Override
      public final void clear() {
          queue.clear();
      }
      @Override
      public final boolean isEmpty() {
          return queue.isEmpty();
      }
  }
  // 具体实现类
  static final class ObserveOnSubscriber<T> extends BaseObserveOnSubscriber<T>
  implements FlowableSubscriber<T> {
      private static final long serialVersionUID = -4547113800637756442L;
      final Subscriber<? super T> downstream;
      ObserveOnSubscriber(
              Subscriber<? super T> actual,
              Worker worker,
              boolean delayError,
              int prefetch) {
          super(worker, delayError, prefetch);
          this.downstream = actual;
      }
      //这是上游回调这个subscriber时调用的方法，详情见上一篇文章
      @Override
      public void onSubscribe(Subscription s) {
          if (SubscriptionHelper.validate(this.upstream, s)) {
              this.upstream = s;
              if (s instanceof QueueSubscription) {
                  @SuppressWarnings("unchecked")
                  QueueSubscription<T> f = (QueueSubscription<T>) s;
                  int m = f.requestFusion(ANY | BOUNDARY);
                  if (m == SYNC) {
                      sourceMode = SYNC;
                      queue = f;
                      done = true;
                      downstream.onSubscribe(this);
                      return;
                  } else
                  if (m == ASYNC) {
                      sourceMode = ASYNC;
                      queue = f;
                      downstream.onSubscribe(this);
                      s.request(prefetch);
                      return;
                  }
              }
              // 设置缓存队列
              // 这里涉及一个特别之处就是预获取（提前获取数据）
              queue = new SpscArrayQueue<T>(prefetch);
              // 触发下游subscriber 如果有request则会触发下游对上游数据的request
              downstream.onSubscribe(this);
              // 请求上游数据 上面的代码和这行代码就是起到承上启下的一个作用，也就是预获取，放在队列中
              s.request(prefetch);
          }
      }
      @Override
      void runSync() {
          int missed = 1;
          final Subscriber<? super T> a = downstream;
          final SimpleQueue<T> q = queue;
          long e = produced;
          for (;;) {
              long r = requested.get();
              while (e != r) {
                  T v;
                  try {
                      v = q.poll();
                  } catch (Throwable ex) {
                      Exceptions.throwIfFatal(ex);
                      cancelled = true;
                      upstream.cancel();
                      a.onError(ex);
                      worker.dispose();
                      return;
                  }
                  if (cancelled) {
                      return;
                  }
                  if (v == null) {
                      cancelled = true;
                      a.onComplete();
                      worker.dispose();
                      return;
                  }
                  a.onNext(v);
                  e++;
              }
              if (cancelled) {
                  return;
              }
              if (q.isEmpty()) {
                  cancelled = true;
                  a.onComplete();
                  worker.dispose();
                  return;
              }
              int w = get();
              if (missed == w) {
                  produced = e;
                  missed = addAndGet(-missed);
                  if (missed == 0) {
                      break;
                  }
              } else {
                  missed = w;
              }
          }
      }
      @Override
      void runAsync() {
          int missed = 1;
          final Subscriber<? super T> a = downstream;
          final SimpleQueue<T> q = queue;
          long e = produced;
          for (;;) {
              long r = requested.get();
              while (e != r) {
                  boolean d = done;
                  T v;
                  try {
                      // 获取数据
                      v = q.poll();
                  } catch (Throwable ex) {
                      Exceptions.throwIfFatal(ex);
                      cancelled = true;
                      upstream.cancel();
                      q.clear();
                      a.onError(ex);
                      worker.dispose();
                      return;
                  }
                  boolean empty = v == null;
                  if (checkTerminated(d, empty, a)) {
                      return;
                  }
                  if (empty) {
                      break;
                  }
                  a.onNext(v);
                  e++;
                  // limit = prefetch - (prefetch >> 2)
                  // prefetch  = BUFFER_SIZE（上一篇文章提到的默认128）
                  // 前面说过，订阅者把自己当成一个发射者，那数/据从哪里来呢，而且还要持续有数据，那么下面代码说明了数据来源，当数据达到limit，开始新的数据的prefetch，每次preftch的数量是limit
                  if (e == limit) {
                      if (r != Long.MAX_VALUE) {
                          r = requested.addAndGet(-e);
                      }
                      upstream.request(e);
                      e = 0L;
                  }
              }
              if (e == r && checkTerminated(done, q.isEmpty(), a)) {
                  return;
              }
              // 下面的代码机制在上一篇讲过主要涉及异步编程技巧
              int w = get();
              if (missed == w) {
                  produced = e;
                  missed = addAndGet(-missed);
                  if (missed == 0) {
                      break;
                  }
              } else {
                  missed = w;
              }
          }
      }
      @Override
      void runBackfused() {
          int missed = 1;
          for (;;) {
              if (cancelled) {
                  return;
              }
              boolean d = done;
              downstream.onNext(null);
              if (d) {
                  cancelled = true;
                  Throwable e = error;
                  if (e != null) {
                      downstream.onError(e);
                  } else {
                      downstream.onComplete();
                  }
                  worker.dispose();
                  return;
              }
              missed = addAndGet(-missed);
              if (missed == 0) {
                  break;
              }
          }
      }
      @Nullable
      @Override
      public T poll() throws Exception {
          T v = queue.poll();
          if (v != null && sourceMode != SYNC) {
              long p = produced + 1;
              if (p == limit) {
                  produced = 0;
                  upstream.request(p);
              } else {
                  produced = p;
              }
          }
          return v;
      }
  }

public final Disposable subscribe(Consumer<? super T> onNext) {
  return subscribe(onNext, Functions.ON_ERROR_MISSING,
      Functions.EMPTY_ACTION, FlowableInternalHelper.RequestMax.INSTANCE);
}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
    Action onComplete, Consumer<? super Subscription> onSubscribe) {
  ObjectHelper.requireNonNull(onNext, "onNext is null");
  ObjectHelper.requireNonNull(onError, "onError is null");
  ObjectHelper.requireNonNull(onComplete, "onComplete is null");
  ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");
  //组装成FlowableSubscriber
  LambdaSubscriber<T> ls = new LambdaSubscriber<T>(onNext, onError, onComplete, onSubscribe);
  //调用核心的订阅方法
  subscribe(ls);
  return ls;
}
public final void subscribe(FlowableSubscriber<? super T> s) {
    ObjectHelper.requireNonNull(s, "s is null");
    try {
        //注册一些钩子这里对此不进行讲解，主要不是核心方法
        Subscriber<? super T> z = RxJavaPlugins.onSubscribe(this, s);
        ObjectHelper.requireNonNull(z, "The RxJavaPlugins.onSubscribe hook returned a null FlowableSubscriber. Please check the handler provided to RxJavaPlugins.setOnFlowableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");
        //核心订阅方法，从名字也能读出是指订阅实际调用处
        //不同的数据产生类也就是实现Flowable抽象类的类
        //比如FlowableCreate，FlowSingle，FlowMap等等去实现自己的实际方法
        subscribeActual(z);
    } catch (NullPointerException e) { // NOPMD
        throw e;
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        // can't call onError because no way to know if a Subscription has been set or not
        // can't call onSubscribe because the call might have set a Subscription already
        RxJavaPlugins.onError(e);
        NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
        npe.initCause(e);
        throw npe;
    }
}
//选择FlowCreate的subscribeActual(Subscriber<? super T> t)方法进行剖析
public void subscribeActual(Subscriber<? super T> t) {
    BaseEmitter<T> emitter;
    //根据不同的回压模式选择不一样的数据发射类
    //神奇的回压模式其实本质上就是一个个数据发射-消费模式
    switch (backpressure) {
    case MISSING: {
        emitter = new MissingEmitter<T>(t);
        break;
    }
    case ERROR: {
        emitter = new ErrorAsyncEmitter<T>(t);
        break;
    }
    case DROP: {
        emitter = new DropAsyncEmitter<T>(t);
        break;
    }
    case LATEST: {
        emitter = new LatestAsyncEmitter<T>(t);
        break;
    }
    default: {
        emitter = new BufferAsyncEmitter<T>(t, bufferSize());
        break;
    }
    }
    //回调注册的FlowableSubscriber的onSubscribe方法
    //这里非常重要，因为这里涉及了rxjava特有的 request请求再消费数据的模式
    //也就是说如果没有request数据，那么就不会调用数据发射（发布）者的onNext方法，
    //那么数据订阅者也就不会消费到数据
    t.onSubscribe(emitter);
    try {
        //回调注册的FlowableOnSubscribe<T> source的subscribe方法
        //这个source其实就是在创建Flow流时注册的数据产生类，进一步验证了上文中
        //提及的其需要实现FlowableOnSubscribe<T>接口
        source.subscribe(emitter);
    } catch (Throwable ex) {
        Exceptions.throwIfFatal(ex);
        emitter.onError(ex);
    }
}
//重点分析BufferAsyncEmitter这个类，看字面意思这是一个switch的默认选择类，
//但其实它是回压策略为BUFFER时的数据发射类
//首先这个类的构造函数具有两个参数，很明显这是 actul就是前面的t这个变量，也就是
//注册的数据消费（订阅）者，capacityHint则是设置容量大小的，默认是128，如果需要扩大需要
//自行设置环境变量 rx2.buffer-size
BufferAsyncEmitter(Subscriber<? super T> actual, int capacityHint) {
    super(actual);
    this.queue = new SpscLinkedArrayQueue<T>(capacityHint);
    this.wip = new AtomicInteger();
}
public void onNext(T t) {
   if (done || isCancelled()) {
       return;
   }
   if (t == null) {
       onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
       return;
   }
   // queue 是存储元素的队列，也就是buffer的核心存储。
   // 当我们开始向下游发送数据的时候首先存入队列，然后下面的drain则是进行核心的
   queue.offer(t);
   drain();
}
//核心的类
void drain() {
  //关键的地方 解决生产速率和消费速率不一致的关键地方，也是我们写并发程序值得借鉴的地方。
  //当数据的产生者（发布）频繁调用onNext方法时，这里产生并发调用关系，wip变量是atomic变量，
  //当第一次执行drain函数时，为0继续执行后面的流程，当快速的继续调用onNext方法时，wip不为0然后返回
  //那么后面的流程我们其实已经很大概率会猜测到应该是去取队列的数据然后做一些操作
 if (wip.getAndIncrement() != 0) {
     return;
 }
 int missed = 1;
 //这里的downstream其实就是注册的数据订阅者，它是基类BaseEmitter的变量，前面初始化时调用了基类的构造函数
 final Subscriber<? super T> a = downstream;
 final SpscLinkedArrayQueue<T> q = queue;
 for (;;) {
     long r = get();
     long e = 0L;
     while (e != r) {
         if (isCancelled()) {
             q.clear();
             return;
         }
         boolean d = done;
         //取队列中的数据
         T o = q.poll();
         boolean empty = o == null;
         if (d && empty) {
             Throwable ex = error;
             if (ex != null) {
                 error(ex);
             } else {
                 complete();
             }
             return;
         }
         if (empty) {
             break;
         }
         //此处回调订阅者的onNext方法去真正的执行数据实例程序
         //到此数据从产生到消费其生命周期已经走完
         a.onNext(o);
         e++;
     }
     if (e == r) {
         if (isCancelled()) {
             q.clear();
             return;
         }
         boolean d = done;
         boolean empty = q.isEmpty();
         if (d && empty) {
             Throwable ex = error;
             if (ex != null) {
                 error(ex);
             } else {
                 complete();
             }
             return;
         }
     }
     if (e != 0) {
        //标记已经消费的个数
         BackpressureHelper.produced(this, e);
     }
     //前面说过wip会原子性的增加，而且是每调用一次onNext增加一次
     //missed从其名解释是指错过的意思，个人理解是错过消费的数据个数，错过消费
     //的意思其实就是指没有进行a.onNext数据消费处理的数据
     missed = wip.addAndGet(-missed);
     if (missed == 0) {
        //如果没有错过的数据也就是全部都消费完那就跳出for循环
        //此处for循环方式和JUC源码中Doug Lea的做法都有类似之处
         break;
        }
    }
}

Disposable disposeable = scheduleObservable
            .map(aLong -> dataAdapter.handlerDpti())
            .map(DataProcess::dataProcess)
            .subscribeOn(Schedulers.single())