RxJava流控机制之Flowable

背景

对于生产者和消费者模型，存在一个问题就是当生产者生产的速度大于消费者消费速度，并且生产过程不会停止，生产者和消费者位于不同的线程中，这是要如何对待多余出来的生产内容？是丢掉，是缓冲？
在强大的异步处理框架中，RxJava又是怎么处理的呢？如果在工作中万一发生丢包事件怎么办？

使用环境与本文目的

RxJava版本：2.1.0
默认条件：观察者和被观察者位于main线程中，且使用了默认的事件发射器。
目的：通过Flowable，探究RxJava的流控机制。

Flowable创建过程

Flowable flowable = Flowable.create(new FlowableOnSubscribe<Integer>() {
                            @Override
                            public void subscribe(FlowableEmitter<Integer> e) throws Exception {
                                for(int i =0 ; i<10;i++){
                                    e.onNext(i);
                                }
                            }
                        }
                , BackpressureStrategy.BUFFER);

在create方法中，完成了对Flowable的构建过程：

public static <T> Flowable<T> create(FlowableOnSubscribe<T> source, BackpressureStrategy mode) {
        ObjectHelper.requireNonNull(source, "source is null");
        ObjectHelper.requireNonNull(mode, "mode is null");
        //在工厂中构建出一个Flowable对象。需要传入对向FlowableCreate
        //如果要构建Observable，则传入的是ObservableDefer
        return RxJavaPlugins.onAssembly(new FlowableCreate<T>(source, mode));
    }

FlowableCreate实际上是Flowable子类。当调用Flowable的subscribe方法时，实际上将执行FlowableCreate中的subscribeActual(该方法在Flowable是一个抽象方法，在FlowableCreate中实现)方法：

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, "Plugin returned null Subscriber");

            subscribeActual(z);
        }
......

subscribe过程分析

实际执行的是subscribeActual，这个方法非常重要，该方法的实现为：

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

        t.onSubscribe(emitter);
        try {
            source.subscribe(emitter);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            emitter.onError(ex);
        }

我们可以看到：

  • 它首先会根据我们选择的背压模式，设置不同的emitter；如果没有设置，默认将开启带有缓存的emitter；
  • Subscriber中的onSubscribe在事件没有发射前就执行了；
  • 事件的发射，是通过source.subscribe(emitter)实现的，而这个source，实际上就是我们在构建Flowable时创建的FlowableOnSubscribe。
    现在回过来我们看看在构建时，FlowableOnSubscribe的内容,通常我们会这么写：
new FlowableOnSubscribe<Integer>() {
                            @Override
                            public void subscribe(FlowableEmitter<Integer> e) throws Exception {
                                for(int i =0 ; i<10;i++){
                                    e.onNext(i);
                                }
                            }
                        }

转了一圈，又回到了这里。FlowableEmitter来发射事件。默认的，将使用BufferAsyncEmitter，这是一个支持背压处理的Emitter。
该Emitter中，onNext方法是这样的：

@Override
        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.offer(t); //生产
            drain();       //实际消费过程会执行queue.poll
        }

我们看到，queue就是它维护的一个SpscLinkedArrayQueue队列(其中使用的大量的原子类型处理多线程访问问题)，队列容量会根据生产消费情况自动扩容。
生产过程，或者说事件发射过程，直接调用了队列的offer方法，进行入队操作；
消费过程，或者说消费事件，则是先使用了drain方法，该方法的本质，是执行队列的poll方法取出事件，然后在onNext()中消费。
在 offer 中主要完成生产：

//producerLookAhead相当于一个生产者的斥候，主要用于检测边界
//这里将检测，要插入的位置，是否已经越界了
 if (index < producerLookAhead) {
            return writeToQueue(buffer, e, index, offset);
        }
//else这种情况，主要时考虑到循环队列
 else {
            //producerLookAheadStep实际上是一个定值，表示固定步长
            final int lookAheadStep = producerLookAheadStep;
            // go around the buffer or resize if full (unless we hit max capacity)
            //首先检查前进了固定步长之后，是否还有位置用来插入，注意，使用calcWrappedOffset方法，
           //包括很多其他用到mask的地方，实际上是将数组作为一个循环队列使用。
           //如果前进固定步长之后，还可以插入，那么，说明生产者可用空间还有很多
            int lookAheadElementOffset = calcWrappedOffset(index + lookAheadStep, mask);
            if (null == lvElement(buffer, lookAheadElementOffset)) { // LoadLoad
                producerLookAhead = index + lookAheadStep - 1; // joy, there's plenty of room
                return writeToQueue(buffer, e, index, offset);
            }
            //检查下一插入位是否为空，如果不为空，则使用 ；
            //反之，插入位已经满了，需要创建一个新的数组以完成生产者的工作
           else if (null == lvElement(buffer, calcWrappedOffset(index + 1, mask))) { // buffer is not full
                return writeToQueue(buffer, e, index, offset);
            } else {
                //现有数组容量已经满了，消费者速度无法跟上生产者速度，
                //需要开辟一块新的空间用于生产。空间大小和现有数组大小一致。
                //这里将完成对已经生产并且尚未消费的数组进行保存的工作；
                //同时，开辟一个新的数组，用于生产
                resize(buffer, index, offset, e, mask); // add a buffer and link old to new
                return true;
            }
}

在poll中主要完成事件取出，以在onNext中消费：

public T poll() {
        // local load of field to avoid repeated loads after volatile reads
        final AtomicReferenceArray<Object> buffer = consumerBuffer;
        final long index = lpConsumerIndex();
        final int mask = consumerMask;
        final int offset = calcWrappedOffset(index, mask);
        final Object e = lvElement(buffer, offset);// LoadLoad
        boolean isNextBuffer = e == HAS_NEXT;
        if (null != e && !isNextBuffer) {
           //取出发射的事件，进行消费
            soElement(buffer, offset, null);// StoreStore
            soConsumerIndex(index + 1);// this ensures correctness on 32bit platforms
            return (T) e;
        } else if (isNextBuffer) {
            //如果这个数组中所有元素已经消费完，同时生产者已经不再这个数组中进行生产工作；
            //跳转到新的数组中，完成消费工作，同时，移除当前数组，即放弃这块空间，不再使用
            return newBufferPoll(lvNext(buffer), index, mask);
        }

        return null;
    }

本质上将，生产和消费就是在操作这样一个队列。
现在，可以回过头来，重新看一看上面的 drain() 方法，看看它具体的消费过程，这个方法很有意思。
drain：

void drain() {
            //保证同时只能有一个线程操作进行下面的循环
            //注意在该方法的末尾，对wip进行了重置为-1，打开进行循环的权限
            if (wip.getAndIncrement() != 0) {
                return;
            }

            int missed = 1;
            final Subscriber<? super T> a = actual;
            //无限队列，本质上有很多个固定长度的数组自动扩展构成
            final SpscLinkedArrayQueue<T> q = queue;
            //死循环
            for (;;) {
                //得到请求的数量，该值是通过Subscription.request()设置的，
                //而这个方法，Subscription参数，实际上在Subscriber的onSubscribe传递进去。
                //所以，这也就是为什么，要在subscriber中request(num), num为多少，就消费多少。
                //这个请求对于外界来说，只能通过subscription设置
                long r = get();
                long e = 0L;
                //如果请求量为0，不进入循环进行消费
                while (e != r) {
                    if (isCancelled()) {
                        q.clear();
                        return;
                    }
                    //用来判断是否执行了onComplete或者onError
                    boolean d = done;

                    T o = q.poll();

                    boolean empty = o == null;
                    //如果事件全部消费完，之后执行了onCopmlete或者onError
                    if (d && empty) {
                        Throwable ex = error;
                        if (ex != null) {
                            super.onError(ex);
                        } else {
                            super.onComplete();
                        }
                        return;
                    }
                    //如果事件全部消费完，跳出本次循环
                    //注意，此时空转了。如果消费者速度大于生产者速度，会发生这次空转，同时继续循环过程
                    if (empty) {
                        break;
                    }
                    //消费事件
                    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) {
                            super.onError(ex);
                        } else {
                            super.onComplete();
                        }
                        return;
                    }
                }
                //上一次request的量已经全部完成，此时重置请求量
                if (e != 0) {
                    BackpressureHelper.produced(this, e);
                }
                //开锁
                missed = wip.addAndGet(-missed);
                if (missed == 0) {
                    break;
                }
            }
        }

总结

上述的分析过程，实际上并没有设置观察者、 被观察者于不同的线程，且使用默认的事件发射器。缓冲队列的空间是无限大的（一旦当前缓冲被使用完，则开辟新的缓冲空间，直到这个空间的容量达到了 long 类型的最大值，或者内存溢出）。
这种背压方式，需要观察者或者消费者主动请求要处理的事件的数量，已达到流速控制。

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容

  • RxJava 2.x 学习2（转载）
    怎么如此平静, 感觉像是走错了片场.为什么呢, 因为上下游工作在同一个线程呀骚年们! 这个时候上游每次调用emit...
    Young1657阅读 1,527评论 2赞 1
  • Spring Cloud
    Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具（例如配置管理，服务发现，断路器，智...
    卡卡罗2017阅读 135,080评论 19赞 139
  • iOS ffmpeg 理解
    教程一:视频截图(Tutorial 01: Making Screencaps) 首先我们需要了解视频文件的一些基...
    90后的思维阅读 4,822评论 0赞 3
  • 源码众享最新教程 - 草稿
    ¥开启¥ 【iAPP实现进入界面执行逐一显】 〖2017-08-25 15:22:14〗 《//首先开一个线程，因...
    小菜c阅读 6,635评论 0赞 17
  • RxJava 2.x 学习（转载）
    转载自：https://xiaobailong24.me/2017/03/18/Android-RxJava2.x...
    Young1657阅读 2,050评论 1赞 9