rxjava3——背压

2018/07/31 2 分钟阅读

简介

简介
Flow Control
背压的使用方式
源码分析
补充

观察者模式有推和拉之分。

推模型

class Subject {
	private List<Observer> observers = new ArrayList<Observer>();
	public void notify(){
		for(Observer observer : observers){
			observer.update(data);	// data 是Subject 全部或部分信息
		}
	}
}

拉模型

class Subject {
	private List<Observer> observers = new ArrayList<Observer>();
	public void notify(){
		for(Observer observer : observers){
			observer.update(this); // observer 看情况通过 Subject 引用获取数据
		}
	}
}

简单说，传统观察者模式的推和拉，就是在观察者的方法参数上弄了点花样，驱动执行的是 Subject（调用Observer 的方法）

而很多文章关于RxJava最友好的文章——背压（Backpressure）在描述rxjava 响应式编程的时候，通常会说：在RxJava的观察者模型中，被观察者是主动的推送数据给观察者，观察者是被动接收的。而响应式拉取则反过来，观察者主动从被观察者那里去拉取数据，而被观察者变成被动的等待通知再发送数据。从这句话看，推和拉的区别在于观察者是否可以控制节奏。

rxjava Backpressure官方文档

Flow Control

本小节主要来自如何形象地描述RxJava中的背压和流控机制？，文章写得太好，以至于只能大段的copy。

从大的方面说，rxjava Backpressure官方文档虽然叫“Backpressure”（背压），但却是在讲述一个更大的话题——“Flow Control”（流控）。Backpressure只是Flow Control的其中一个方案。

Flow Control有哪些思路呢？大概是有四种:

背压（Backpressure）。
节流（Throttling）。
打包处理。
调用栈阻塞（Callstack blocking）。

Backpressure

Backpressure，也称为Reactive Pull，就是下游需要多少（具体是通过下游的request请求指定需要多少），上游就发送多少。这有点类似于TCP里的流量控制，接收方根据自己的接收窗口的情况来控制接收速率，并通过反向的ACK包来控制发送方的发送速率。

这种方案只对于所谓的cold Observable有效。cold Observable指的是那些允许降低速率的发送源，比如两台机器传一个文件，速率可大可小，即使降低到每秒几个字节，只要时间足够长，还是能够完成的。相反的例子是音视频直播，数据速率低于某个值整个功能就没法用了（这种就属于hot Observable了）。

节流（Throttling）

节流（Throttling），说白了就是丢弃。消费不过来，就处理其中一部分，剩下的丢弃。

而至于处理哪些和丢弃哪些数据，就有不同的策略。主要有三种策略：

sample (也叫throttleLast)
throttleFirst
debounce (也叫throttleWithTimeout)

打包处理

打包就是把上游来的小包裹打成大包裹，分发到下游。这样下游需要处理的包裹的个数就减少了。RxJava中提供了两类这样的机制：buffer和window。buffer和window的功能基本一样，只是输出格式不太一样：buffer打包后的包裹用一个List表示，而window打包后的包裹又是一个Observable。

调用栈阻塞（Callstack blocking）

这是一种特殊情况，阻塞住整个调用栈（Callstack blocking）。之所以说这是一种特殊情况，是因为这种方式只适用于整个调用链都在一个线程上同步执行的情况，这要求中间的各个operator都不能启动新的线程。

“调用栈阻塞”相当于很多车行驶在盘山公路上，而公路只有一条车道。那么排在最前面的第一辆车就挡住了整条路，后面的车也只能排在后面。而“Backpressure”相当于银行办业务时的窗口叫号，窗口主动叫某个号过去（相当于请求），那个人才过去办理。

背压的使用方式

关于RxJava最友好的文章——背压（Backpressure）

Observable observable=Observable.range(1,100000);
class MySubscriber extends Subscriber<T> {
    @Override
    public void onStart() {
    //一定要在onStart中通知被观察者先发送一个事件
      request(1);
    }
    @Override
    public void onCompleted() {
        ...
    }
    @Override
    public void onError(Throwable e) {
        ...
    }
    @Override
    public void onNext(T n) {
        ...
        ...
        //处理完毕之后，在通知被观察者发送下一个事件
        request(1);
    }
}
observable.observeOn(Schedulers.newThread())
            .subscribe(MySubscriber);

背压（Backpressure）并不是一个像flatMap一样可以在程序中直接使用的操作符。上文例子中，最先调用Subscriber的onStart方法，执行request(1) 告诉Observable 发送一个数据，触发Subscriber 执行一次onNext，然后在onNext结束时执行request(1)，以此循环。

源码分析

rxjava 最早出现 Backpressure 是在 0.20.0-RC1，本文分析其稳定版 0.20.0

新引入一个Producer 接口

public interface Producer {
    public void request(long n);
}

Subscriber 加了新方法

public abstract class Subscriber<T> implements Observer<T>, Subscription {
    public void onStart();
    protected final void request(long n);
    public final void setProducer(Producer producer);
}

那他们如何相互作用呢？以Iterable 的实现为例

0.20.0 版本

public final static <T> Observable<T> from(Iterable<? extends T> iterable) {
    return create(new OnSubscribeFromIterable<T>(iterable));
}

public final class OnSubscribeFromIterable<T> implements OnSubscribe<T> {
    final Iterable<? extends T> is;
    public OnSubscribeFromIterable(Iterable<? extends T> iterable) {
        this.is = iterable;
    }
    @Override
    public void call(final Subscriber<? super T> o) {
        if (is == null) {
            o.onCompleted();
        }
        final Iterator<? extends T> it = is.iterator();
        o.setProducer(new IterableProducer<T>(o, it));
    }
}

IterableProducer 忽略性能和安全方面的代码，简化如下

private static final class IterableProducer<T> implements Producer {
	private final Subscriber<? super T> o;
    private final Iterator<? extends T> it;
    public void request(long n) {
        int _c = requested.getAndAdd(n);
        if (_c == 0) {
            while (it.hasNext()) {
                if (o.isUnsubscribed()) {
                    return;
                }
                T t = it.next();
                o.onNext(t);
                if (requested.decrementAndGet() == 0) {
                    // we're done emitting the number requested so return
                    return;
                }
            }
            o.onCompleted();
        }
    }
}

0.10.0 版本，不支持背压

public static <T> Observable<T> toObservable(Iterable<T> iterable) {
    return create(OperationToObservableIterable.toObservableIterable(iterable));
}

public final class OperationToObservableIterable<T> {
    public static <T> Func1<Observer<T>, Subscription> toObservableIterable(Iterable<T> list) {
        return new ToObservableIterable<T>(list);
    }
    private static class ToObservableIterable<T> implements Func1<Observer<T>, Subscription> {
        public ToObservableIterable(Iterable<T> list) {
            this.iterable = list;
        }
        public Iterable<T> iterable;
        public Subscription call(Observer<T> observer) {
            for (T item : iterable) {
                observer.onNext(item);
            }
            observer.onCompleted();
            return Subscriptions.empty();
        }
    }
}

可以看到

Subscriber/Observer 通过producer ，持有了数据源 Iterator
0.20.0 OnSubscribeFromIterable 的OnSubscribe.call 只是为 Observer 设置了producer 成员。而0.10.0 版本的OperationToObservableIterable.call 实际触发了 (所有)observer.onNext 的执行。

所以对比 rxjava2——线程切换提出的方法执行链条分析的方式，0.20.0 版本的执行链条为

 Observable4.subscribe ==>  
 Observable4.onSubscribe.call ==> 	
 Observable3.subscribe ==> 
 Observable3.subscribeOn.call ==> 
 Observable2.subscribe ==> 
 Observable2.onSubscribe.call ==> 
 Observable1.subscribe ==> 
 Observable1.onSubscribe.call ==>
 Observable.subscribe ==> 
 Observable.onSubscribe.call ==> 
 observer4.onNext ==> 
 filter ==> 
 observer3.onNext ==> 
 map ==>
 observer2.onNext ==> // 提交事件，驱动线程执行完毕，另一个线程执行下面的逻辑（接收事件并驱动后续执行）
	
 ...
 observer.onNext
 request(1) ==>
 producer.request(1) ==>
 observer4.onNext  第二轮
 ...

可以看到，这个链条就是一个典型的拉模型，所谓响应式拉取：观察者主动从被观察者那里去拉取数据，而被观察者变成被动的等待通知再发送数据。终于实至名归了。
rxjava2——线程切换笔者看的是0.10.0 的代码，从当时的执行链条看，rxjava 也是一个推模型：被观察者是主动的推送数据给观察者。彼时还困惑不已，以为博客们都说错了，现在可以看到，0.10.0 到 0.20.0 经历一个理念的变化。

补充

关于RxJava最友好的文章——背压（Backpressure）中着重提到背压策略的一个前提是异步环境，也就是说，被观察者和观察者处在不同的线程环境中。否则，在一个线程中，区分推和拉，观察者和被观察者谁做主导，谁驱动逻辑，就没意思了。

李乾坤

rxjava3——背压

简介

Flow Control

Backpressure

节流（Throttling）

打包处理

调用栈阻塞（Callstack blocking）

背压的使用方式

源码分析

0.20.0 版本

0.10.0 版本，不支持背压

可以看到

补充

留下评论

猜您还喜欢

Long-horizon Agent

next prompt工程——skill

从chatbot到clawbot

typescript学习