Buffer操作符功能介绍
如果本来是Observable<String>的一个对象，那么我只能在onNext中一个一个处理String,通过Buffer操作符，我们可以把n个String组合在一起，然后在onNext中进行处理
用途
可以操作一些需要一起处理的数据。比如2个数据为一组，我始终要打印最大的数据，或者3个数据为一组，打印最大值。
来一段代码
btSub.setOnClickListener({
observable?.subscribe({ msg ->
tvContent.text = tvContent.text.toString() + "\n" + msg.toString()
})
})
observable = Observable.just("test1","test2","test3").buffer(2)
代码的作用就是每2个字符串 打印在一行文本中。
先来学习下just是什么鬼
Observable.create(object : Observable.OnSubscribe<String> {
override fun call(t: Subscriber<in String>) {
t.onNext("test1")
t.onNext("test2")
t.onNext("test3")
}
})
....
Observable.just("test1","test2","test3")
这2行代码的效果其实是一致的(当我们，深入源代码的时候，其实会发现有一些不同，但是我们现在先一个个学习操作符，然后再会过来看这些，其实会非常的简单)
这里我就不对just进行深入了，主要以操作符为主，如果感兴趣的可以自己看看。
看看源代码
Observable
...
public final Observable<List<T>> buffer(int count) {
return buffer(count, count);
}
...
public final Observable<List<T>> buffer(int count, int skip) {
return lift(new OperatorBufferWithSize<T>(count, skip));
}
...
public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
return unsafeCreate(new OnSubscribeLift<T, R>(onSubscribe, operator));
}
...
这里我们看到了一个很牛逼的东西，也就是lift，在前言中，其实我也给大家推荐了一篇关于RxJava的文章，写的很经典，大家最好先看看。
既然提到了lift那么就不得不提到Operator接口了。前面其实我也已经介绍过，RxJava中最重要的4个类或接口，如果忘了，或者不太了解的，大家可以再去看下那
篇文章我是最简单的操作符-----Create
public interface Operator<R, T> extends Func1<Subscriber<? super R>, Subscriber<? super T>> {
// cover for generics insanity
}
public interface Func1<T, R> extends Function {
R call(T t);
}
这段代码大家仔细一看就发现，其实Operator也是一个Func1嘛，只有一个call接口，用来把一个Subscriber转化成了另外一个Subscriber。
既然已经了解了这Operator，那么我们继续看lift到底做了什么。一步步看代码。
前面我介绍了一种方法去阅读源代码，就是我们已经知道了它的功能，就带着他这个功能去阅读源码，这里也是适用的。
以我开头的demo代码为例子，其实就是把Subscriber<List<String>>转化为了Subscriber<String>
这里其实我自己理解起来都非常困难，为什么是把Subscriber<List<String>>转化为了Subscriber<String>而不是把Subscriber<String>转化为了Subscriber<List<String>>
先不解释，先直接看OperatorBufferWithSize的代码
OperatorBufferWithSize
...
public Subscriber<? super T> call(final Subscriber<? super List<T>> child) {
...
}
...
因为我们最习惯的思考逻辑就是从上到下，从左到右，因为一开始我们是Test1，Test2，Test3，然后呢组成，[Test1，Test2]，[Test3]这样子去处理，所以很正常的理解为把Subscriber<String>转化为了Subscriber<List<String>>。这也是为什么我觉得RxJava源码看起来比较累的原因。
在前言中介绍的那篇经典的文章中就有一幅图，我就不引用了，大家可以自己去看一下，RxJava的运行顺序是先下再上，再从上到下。我们直接继续从源码中找寻答案把。
Observable
public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
return unsafeCreate(new OnSubscribeLift<T, R>(onSubscribe, operator));
}
OnSubscribeLift
...
public void call(Subscriber<? super R> o) {
try {
Subscriber<? super T> st = RxJavaHooks.onObservableLift(operator).call(o);
try {
st.onStart();
parent.call(st);
}
...
}
RxJavaHooks我就不过多介绍了，在前几篇文章中已经有所解释，所以以上代码简化为
...
public void call(Subscriber<? super R> o) {
try {
Subscriber<? super T> st = operator.call(o);
try {
st.onStart();
parent.call(st);
}
...
}
当我们点击demo中的按钮，就是直接先调用上面的call方法。这里的parent其实是Observable中传入的onSubscribe，其实也就是前面所提到的
Observable.OnSubscribe<String> {
override fun call(t: Subscriber<in String>) {
t.onNext("test1")
t.onNext("test2")
t.onNext("test3")
}
}
所以直接看到这里，其实大家应该可以看出来为什么是把Subscriber<List<String>>转化为Subscriber<String>了。大家再来重新看这段代码。
OnSubscribeLift
...
public void call(Subscriber<? super R> o) {
try {
Subscriber<? super T> st = operator.call(o);
try {
st.onStart();
parent.call(st);
}
...
这里的参数Subscriber<? super R> o，就是我们在demo中subscribe订阅方法中传入的Subscriber。所以R其实就是List<String>，而在call方法中最终调用的是parent.call所传的参数是Subscriber<String>。调用的是下面这个方法
Observable.OnSubscribe<String> {
override fun call(t: Subscriber<in String>) {
t.onNext("test1")
t.onNext("test2")
t.onNext("test3")
}
}
所以我说，
RxJava的执行方式呢，最后面的OnSubscribe的call先执行，然后再调用前面的OnSubscribe，最前面的OnSubscribe开始调用onNext方法，再一步一步往下执行。
这里我们对整个方法的执行流程有了一个大致的了解，但是还没有开始去解析buffer操作符的真正实现的代码。
下面我们还是继续看
OperatorBufferWithSize
...
public Subscriber<? super T> call(final Subscriber<? super List<T>> child) {
if (skip == count) {
BufferExact<T> parent = new BufferExact<T>(child, count);
...
return parent;
}
if (skip > count) {
BufferSkip<T> parent = new BufferSkip<T>(child, count, skip);
...
return parent;
}
BufferOverlap<T> parent = new BufferOverlap<T>(child, count, skip);
...
return parent;
}
...
这里非常明显的分为了3中情况，skip代表是忽略几个元素，count代表几个元素为1组。什么意思呢？
我们这里直接举个例子。
Observable.just("Test1","Test2","Test3","Test4","Test5")
元素多一点，大家好理解，那么我skip=2,count=2,那么就会被分为3组，"[Test1,Test2]","[Test3,Test4]","[Test5]"。如果skip=3,count=2,那么就会被分为2组，"[Test1,Test2]"，"[Test4,Test5]"。
大家自己尝试下，我就不举例了，我就想说明，这里分了3种情况做了不同处理，其实没什么关系，我们只需看明白一种就好了，其他2种也是大同小异的。我们的demo中
observable = Observable.just("test1", "test2", "test3").buffer(2)只传了一个2，就是skip==2&&count==2。
所以我们直接看
...
if (skip == count) {
BufferExact<T> parent = new BufferExact<T>(child, count);
...
return parent;
}
...
static final class BufferExact<T> extends Subscriber<T> {
final Subscriber<? super List<T>> actual;
final int count;
List<T> buffer;
public BufferExact(Subscriber<? super List<T>> actual, int count) {
this.actual = actual;
this.count = count;
this.request(0L);
}
@Override
public void onNext(T t) {
List<T> b = buffer;
if (b == null) {
b = new ArrayList<T>(count);
buffer = b;
}
b.add(t);
if (b.size() == count) {
buffer = null;
actual.onNext(b);
}
}
...
前面介绍了Operator就是为了把 Subscriber<List<String>>转为 Subscriber<String>，而这里的BufferExact就是这个Subscriber<String>。
所以我们最主要的还是看onNext的实现。
public void onNext(T t) {
List<T> b = buffer;
if (b == null) {
b = new ArrayList<T>(count);
buffer = b;
}
b.add(t);
if (b.size() == count) {
buffer = null;
actual.onNext(b);
}
}
实现非常的简单，先把onNext传递过来的数据先保存在 List中，等凑够了count数量后，在调用后面的Subscriber<List<String>>。
所以总的来说
buffer操作符的实现是简单的，但是由于涉及到了lift和Operator所以整体看起来还是比较复杂的。大家可以自己琢磨。
附加
前面直接说
Observable.create(object : Observable.OnSubscribe<String> {
override fun call(t: Subscriber<in String>) {
t.onNext("test1")
t.onNext("test2")
t.onNext("test3")
}
})
Observable.just("test1", "test2", "test3")
这2个是差不多的。其实我们可以再深入下。
Observable
...
public static <T> Observable<T> just(T t1, T t2, T t3) {
return from((T[])new Object[] { t1, t2, t3 });
}
...
public static <T> Observable<T> from(T[] array) {
int n = array.length;
if (n == 0) {
return empty();
} else
if (n == 1) {
return just(array[0]);
}
return unsafeCreate(new OnSubscribeFromArray<T>(array));
}
...
其实就是创建了一个OnSubscribeFromArray，也就是一个OnSubscribe。继续深入
OnSubscribeFromArray
...
public void call(Subscriber<? super T> child) {
child.setProducer(new FromArrayProducer<T>(child, array));
}
..
static final class FromArrayProducer<T>
extends AtomicLong
implements Producer {
...
@Override
public void request(long n) {
if (n < 0) {
throw new IllegalArgumentException("n >= 0 required but it was " + n);
}
if (n == Long.MAX_VALUE) {
if (BackpressureUtils.getAndAddRequest(this, n) == 0) {
fastPath();
}
} else
if (n != 0) {
if (BackpressureUtils.getAndAddRequest(this, n) == 0) {
slowPath(n);
}
}
}
...
如果是OnSubscribe其实我们主要就是看一个call方法，这里是直接child.setProducer，前面也介绍过，setProducer其实无法是直接调用Producer的request的方法。最后我们发现其实主要就看2个方法fastPath和slowPath。
...
void fastPath() {
final Subscriber<? super T> child = this.child;
for (T t : array) {
if (child.isUnsubscribed()) {
return;
}
child.onNext(t);
}
if (child.isUnsubscribed()) {
return;
}
child.onCompleted();
}
...
void slowPath(long r) {
final Subscriber<? super T> child = this.child;
final T[] array = this.array;
final int n = array.length;
long e = 0L;
int i = index;
for (;;) {
while (r != 0L && i != n) {
if (child.isUnsubscribed()) {
return;
}
child.onNext(array[i]);
i++;
if (i == n) {
if (!child.isUnsubscribed()) {
child.onCompleted();
}
return;
}
r--;
e--;
}
r = get() + e;
if (r == 0L) {
index = i;
r = addAndGet(e);
if (r == 0L) {
return;
}
e = 0L;
}
}
}
...
其实直接看fastPath会更加清爽一点，就是遍历数组，然后挨个调用onNext,那么数组是什么呢，就是前面传入的"test1", "test2", "test3"。
至于slowPath呢其实关键代码也是onNext。可以先不深入去理解它具体的意思，其实看名字就可以看出来，一个是快速处理，一个是慢慢处理。
总结
整体下来我也已经尽量涉及到了每行代码，但是中间的跳转确实会让很多人发懵，如果觉得哪里讲的还不够清楚，可以指出，我可以修正。
网友评论