RxJava使用takeUntil操作实现生命周期的绑定

相信使用过RxJava、RxAndroid的人都会被其简便的流式操作所吸引，结合java8的lambda表达式，使代码看起来更简洁，操作逻辑更清晰，数据处理更便捷，告别了臃肿的代码。但要优雅的使用它还需要进行一些处理，避免操作不当导致的内存泄漏风险。

---

管理订阅的几种方式

• 使用CompositeSubscription统一管理生命周期(后面具体讲解实现方式)
• 使用RxLifecycle来管理订阅，实现和Android组件的生命周期绑定。
• 使用takeUntil操作符结合BehaviorSubject实现生命周期绑定。

---

具体介绍

一、使用CompositeSubscription统一管理生命周期

public class BaseActivity extends AppCompatActivity {    
    protected CompositeSubscription subscription = new   CompositeSubscription();    
    protected void addSub(Subscription sub) {        
        if (sub != null && !sub.isUnsubscribed()) {                   
          subscription.add(sub);       
     }   
 }    
@Override   
protected void onDestroy() {     
     if(subscription.hasSubscriptions()){          
          subscription.unsubscribe();        
    }        
    super.onDestroy();   
 }
}

在子类统一调用addSub方法将生成的订阅关系添加进subscription管理类中，在onDestroy中统一取消订阅关系。

二、使用RxLifecycle来管理订阅(需要引入新的类库)

这里不进行详细描述，具体操作参照Rxlifecycle使用详解

三、使用takeUntil操作符结合BehaviorSubject实现生命周期绑定

具体代码实现：

//这里提供的是kotlin的实现
class MainActivity : AppCompatActivity() {    
  var subject: BehaviorSubject<Event> = BehaviorSubject.create()    
  val TAG = MainActivity::class.java!!.simpleName    
  override fun onCreate(@Nullable savedInstanceState: Bundle?) {   
              super.onCreate(savedInstanceState)        
              setContentView(R.layout.activity_main)       
     var sub = Observable.interval(1, TimeUnit.SECONDS)
    .compose(this.bindLife<Long>())
    .subscribe({ a -> Log.i(TAG, "call: " + a!!) },
    { t -> Log.e(TAG, "call: " + t.message!!) }                       
   , { Log.i(TAG, "complete: 完成") })
}    

  fun <T> bindLife(): Observable.Transformer<T, T> { 
    return Observable.Transformer<T, T> {           
     observable -> observable.takeUntil(subject.skipWhile(
    {event ->event != Event.PAUSE && event != Event.DESTROY
    && event != Event.DETACH }))}    
  }   
 
  override fun onDestroy() { 
    subject.onNext(Event.DESTROY)
    super.onDestroy()    
  }
}

enum class Event {    
// Activity life Events    
  CREATE,   
  START,  
  RESUME,  
  PAUSE,  
  STOP,   
  DESTROY,    
// Fragment life  Events  
  ATTACH, 
  CREATE_VIEW,  
  DESTROY_VIEW,  
  DETACH
}

java版本(非lambda)

/**
 * ClassName: BaseActivity<p>
 * Author: Alpha<p>
 * Fuction: <p>
 * CreateDate: 2016/7/3 0:17<p>
 */
public class BaseActivity extends AppCompatActivity {

    protected BehaviorSubject<Event> subject = BehaviorSubject.create();

    @Override
    protected void onCreate(@Nullable Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        Observable.interval(3, TimeUnit.SECONDS)
                .compose(this.<Long>bindLife())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<Long>() {
                    @Override
                    public void onCompleted() {
                        Log.i("rxjava", "onCompleted");

                        Toast.makeText(BaseActivity.this, "onCompleted", Toast.LENGTH_SHORT).show();

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(final Long aLong) {
                        Log.i("rxjava", "onNext " + aLong);

                        Toast.makeText(BaseActivity.this, "aLong:" + aLong, Toast.LENGTH_SHORT).show();
                    }
                });
    }

    protected <T> Observable.Transformer<T, T> bindLife() {
        return new Observable.Transformer<T, T>() {
            @Override
            public Observable<T> call(Observable<T> observable) {
                return observable.takeUntil(subject.skipWhile(new Func1<Event, Boolean>() {
                    @Override
                    public Boolean call(Event event) {
                        return event != Event.DESTROY && event != Event.DETACH;
                    }
                }));
            }
        };
    }

    @Override
    protected void onDestroy() {
        subject.onNext(Event.DESTROY);
        super.onDestroy();
    }
}

未处理之前.gif 处理之后.gif

鸣谢：#Kotlin# 小心 Rx 的生命周期

感谢大家的支持，如有不足之处希望大家提出来，一起交流进步。