美文网首页
RXJava Subscriber从源码中分析不能被重用的原因

RXJava Subscriber从源码中分析不能被重用的原因

作者: heybik | 来源:发表于2016-05-15 18:54 被阅读1758次

RxJava是android中比较流行的框架,在刚开始使用这个框架时发现一个奇怪的问题,现在解决了,写下来;

我们都知道RxJava主要分为Observable and Observer; 其中的Observer有一个具体的实现类 Subscriber,所以在开发中我们经常会使用Subscriber来代替Observer,大部分时候这二者是一样的,

由于Subscriber还实现了Subscription 所以比Observer要多2个方法;

1.正常的写法如下:

Observable<String> mObservable;
Subscriber<String> mSubscriber;
private Subscriber<String> createRXSubscriber(){ 
Subscriber<String> stringSubscriber = new Subscriber<String>() {        
   @Override        
   public void onCompleted() {            
     Log.d(TAG,"Subscriber->onCompleted");       
   }      
   @Override       
   public void onError(Throwable e) {
     Log.d(TAG,"Subscriber->onError");       
   }        
   @Override        
   public void onNext(String s) {           
     Log.d(TAG,"Subscriber->onNext:"+s);        
     //textV.setText(s);        
   }    
};    

return stringSubscriber;
}
private Observable<String> createRXObserable(){    
  return Observable.create(new Observable.OnSubscribe<String>(){        
  @Override        
   public void call(Subscriber<? super String> subscriber) {        
        subscriber.onNext("this is test for Rxjava!");    
        subscriber.onCompleted();    
   }   
 });
}
mObservable = createRXObserable();
mSubscriber = createRXSubscriber();
btn.setOnClickListener(new View.OnClickListener() {  
  @Override    
  public void onClick(View view) {        
    mObserable.subscribe(mSubscriber);       
  }
});

这样无论我们的btn被点击了几次都是看到mSubscriber的onNext 执行打印的log :this is test for Rxjava!

2.当使用Observable提供的just(...) ,from(...) 方法来实现的时候

private Observable<String> createRXObserable2(){    
  return Observable.just("Hello", "RX", "java");
}
private Observable<String> createRXObserable3(String[] events){    
  return Observable.from(events);
}
mObservable = createRXObserable2();
//mObservable = createRXObserable3(new String[]{"hello","RX","sample"});
mSubscriber = createRXSubscriber();
btn.setOnClickListener(new View.OnClickListener() {  
  @Override    
  public void onClick(View view) {        
    mObserable.subscribe(mSubscriber);       
  }
});

当点击btn后会发现除了第一次点击,以后的每次都不会打印出我们的相应的log出来,即Subscriber并没有再按预期执行了;
原因是由于在源码内部中会最终在事件结束后调用Subscriber的onCompleted(),或者是发生了错误不调用onError;(准确的说是我们传入的Subscriber会被包装成一个SafeSubscriber 对象然后这个对象的其中一个属性actual是我们自己new的Subscriber对象);

// new Subscriber so onStart itsubscriber.onStart();/* * See https://github.com/ReactiveX/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls * to user code from within an Observer" */// if not already wrapped
if (!(subscriber instanceof SafeSubscriber)) {   
 // assign to `observer` so we return the protected version    
subscriber = new SafeSubscriber<T>(subscriber);
}

而SafeSubscriber中当执行了onCompleted(),onError()后主动调用unsubscribe();

@Override
public void onCompleted() {   
 if (!done) {       
 done = true;       
 try {            
actual.onCompleted();       
 } catch (Throwable e) {           
 // we handle here instead of another method so we don't add stacks to the frame            
// which can prevent it from being able to handle StackOverflow            
Exceptions.throwIfFatal(e);            RxJavaPluginUtils.handleException(e);           
 throw new OnCompletedFailedException(e.getMessage(), e);   
     } finally {         
   try {               
 // Similarly to onError if failure occurs in unsubscribe then Rx contract is broken                
// and we throw an UnsubscribeFailureException.  
   unsubscribe();            
} catch (Throwable e) {        
        RxJavaPluginUtils.handleException(e);               
 throw new UnsubscribeFailedException(e.getMessage(), e);  
          }      
  }    
}}

针对just(),from()这种方式生成事件方式参考源码中:

public final class OnSubscribeFromArray<T> implements OnSubscribe<T> {    
final T[] array;    
public OnSubscribeFromArray(T[] array) {        
this.array = array;    
}        
@Override    
public void call(Subscriber<? super T> child) {        
child.setProducer(new FromArrayProducer<T>(child, array));   
 }
...

最后在call被调用的时候会去执行setProducer()...这其中就会检查当前是否是isUnSubscribred() 具体代码就不再贴了,如果是就不会执行onNext(),onCompleted()了;

综上所述当使用just(),from这种方式生成事件时,当事件发生之后会执行onCompleted(),而这个方法中就会进行unsubscribe(),取消订阅,当取消了订阅就是不可逆的,所以以后每次发生事件都会因为判断是取消订阅了 所以无法再执行onNext()...;官方给出的建议是每次使用都new一个新的Subscriber 或者是使用Observer;


另为什么第一种通过create创建的Observable 就可以重复使用一个固定的Subscriber呢?其实并不复杂,因为我们重写了OnSubscribe中的call()方法,在里面直接去call onNext(),onCompleted();这样我们不会如上面一样去执行setProducer()...也就不会有去检查是否被取消订阅了;


这是我在使用Rxjava中遇到的问题,如写的有问题还请指正;
在使用一个框架的时候,只需要关注接口api就行了 如果要知道问题的前因后果的话最好还是追一追源码;

相关文章

网友评论

      本文标题:RXJava Subscriber从源码中分析不能被重用的原因

      本文链接:https://www.haomeiwen.com/subject/ofbvrttx.html