深入RxJava2 源码解析(一)

// 创建flowable
Flowable<Map<String, Map<String,Object>>> esFlowable = Flowable.create(new ElasticSearchAdapter(), BackpressureStrategy.BUFFER);
Disposable disposeable = esFlowable
    // map操作 1.采集、2.清洗
    .map(DataProcess::dataProcess)
    .subscribeOn(Schedulers.single())
    //计算任务调度器
    .observeOn(Schedulers.computation())
    // 订阅者 consumer 执行运算
    .subscribe(keyMaps -> new PredictEntranceForkJoin().predictLogic(keyMaps));

FlowableOnSubscribe<T> source

void subscribe(@NonNull FlowableEmitter<T> emitter) throws Exception

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;
    }
}

public void subscribeActual(Subscriber<? super T> t) {
    BaseEmitter<T> emitter;
    //根据不同的回压模式选择不一样的数据发射类
    //神奇的回压模式其实本质上就是一个个数据发射-消费模式
    switch (backpressure) {
    case MISSING: {
        emitter = new MissingEmitter<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())

public final <R> Flowable<R> map(Function<? super T, ? extends R> mapper) {
        ObjectHelper.requireNonNull(mapper, "mapper is null");
        return RxJavaPlugins.onAssembly(new FlowableMap<T, R>(this, mapper));
    }
public final Flowable<T> subscribeOn(@NonNull Scheduler scheduler, boolean requestOn) {
        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
        return RxJavaPlugins.onAssembly(new FlowableSubscribeOn<T>(this, scheduler, requestOn));
}

//很明显 实现的抽象类其实是装修者抽象类
public final class FlowableSubscribeOn<T> extends AbstractFlowableWithUpstream<T , T>

// 这个在前面我们重点分析过这是实际订阅执行的类方法，其实也就是我们说的装饰方法，里面实现了每个类自己的特定“装修”方法
@Override
public void subscribeActual(final Subscriber<? super T> s) {
    // 获取订阅者，下一篇文章会重点讲述rxjava的线程池分配机制
    Scheduler.Worker w = scheduler.createWorker();
    final SubscribeOnSubscriber<T> sos = new SubscribeOnSubscriber<T>(s, w, source, nonScheduledRequests);
    // 跟前面一样调用数据订阅者的onSubscribe方法
    s.onSubscribe(sos);
    // 由分配的调度者进行订阅任务的执行
    w.schedule(sos);
}

// 开始分析SubscribeOnSubscriber这个静态内部类的内部代码
// 实现了Runable用来异步执行
static final class SubscribeOnSubscriber<T> extends AtomicReference<Thread>
  implements FlowableSubscriber<T>, Subscription, Runnable
  // 下游订阅引用
  final Subscriber<? super T> downstream;
  // 上游发射类引用
  final AtomicReference<Subscription> upstream;
  // 上游数据源引用 跟上游引用有区别，简单的说每个上游数据源引用有自己的上游发射类
  Publisher<T> source;
// 这里是装饰的核心代码
@Override
public void run() {
    lazySet(Thread.currentThread());
    // source即为上游，表示其所装饰的源
    Publisher<T> src = source;
    source = null;
    // 调用上游的自身的subscribe方法，在上面一开始我们说这个方法内部会去调用自身实现的subscribeActual方法
    // 从而实现上游自己的特定方法，比如假设source是FlowCreate那么此处就会调用前面一开始我们所讲到的数据的发射
    src.subscribe(this);
}

// 既然已经保证了数据的发射那么数据的处理是不是也要处理
// 很明显这是调用了下游订阅者的onNext方法
@Override
public void onNext(T t) {
    downstream.onNext(t);
}