android使用RxJava实现预加载

在中介绍了使用非rxjava环境下，使用handler机制syncbarrier的特性实现预加载功能的方法。

在rxjava的环境下使用behaviorsubject的特性来实现也是很方便的。

behaviorsubject内部会缓存消息流中最近的一个消息, 在后续有subscriber订阅时，会直接将缓存的消息发送给subscriber。

rxpreloader.java封装如下：

import android.support.annotation.nonnull;

import java.util.linkedlist;

import rx.observable;
import rx.observer;
import rx.subscriber;
import rx.subscription;
import rx.android.schedulers.androidschedulers;
import rx.functions.action1;
import rx.schedulers.schedulers;
import rx.subjects.behaviorsubject;

/**
 * 预加载
 * preloader = rxpreloader.preload(observable);
 * preloader.get(observer1);
 * preloader.get(observer2);
 * preloader.reload();
 * preloader.destroy()
 *
 * @author billy.qi
 */
public class rxpreloader<t> {
 private behaviorsubject<t> subject;
 private observable<t> observable;
 private subscription subscription;
 private final linkedlist<subscription> allobserver = new linkedlist<>();


 private rxpreloader(observable<t> observable) {
 //注意的是由于oncompleted也是数据流中的一个
 //如果直接observer.subscribeon(schedulers.io()).subscribe(subject);
 //会导致subject只能缓存oncompleted
 //所以此处新建一个onsubscribe，通过调用subject.onnext(t)的方式来缓存数据
 this.observable = observable;
 subject = behaviorsubject.create();
 subscription = observable.create(new observable.onsubscribe<t>() {
  @override
  public void call(subscriber<? super t> subscriber) {
  performload();
  }
 })
 .subscribeon(schedulers.io())
 .subscribe(subject);
 }

 public static <r> rxpreloader<r> preload(@nonnull observable<r> observable) {
 return new rxpreloader<r>(observable);
 }

 public void reload() {
 performload();
 }

 public subscription get(observer<t> observer) {
 subscription subscription = subject.observeon(androidschedulers.mainthread())
  .subscribe(observer);
 allobserver.add(subscription);
 return subscription;
 }


 private void performload() {
 observable.subscribeon(schedulers.io())
  .subscribe(new action1<t>() {
   @override
   public void call(t t) {
   if (subject != null) {
    subject.onnext(t);
   }
   }
  }, new action1<throwable>() {
   @override
   public void call(throwable throwable) {
   throwable.printstacktrace();
   }
  });
 }

 public void destroy() {
 synchronized (allobserver) {
  while(!allobserver.isempty()) {
  unsubscribe(allobserver.removefirst());
  }
 }
 unsubscribe(subscription);
 subscription = null;
 subject = null;
 }

 private void unsubscribe(subscription subscription) {
 if (subscription != null && !subscription.isunsubscribed()) {
  subscription.unsubscribe();
 }
 }
}

在activity中使用：

import android.os.bundle;
import android.support.v7.app.appcompatactivity;
import android.widget.textview;

import java.util.concurrent.timeunit;

import rx.observable;
import rx.subscriber;

/**
 * 使用rxjava实现的预加载方式
 */
public class rxpreloaderactivity extends appcompatactivity {

 private textview textview;
 private rxpreloader<string> preloader;

 @override
 protected void oncreate(bundle savedinstancestate) {
 preload();//启动预加载
 initlayout(savedinstancestate);
 preloader.get(observer);//展示预加载的数据
 }

 //初始化布局
 private void initlayout(bundle savedinstancestate) {
 super.oncreate(savedinstancestate);
 setcontentview(r.layout.activity_main);
 settitle("使用rxpreloader");
 //通过循环多次findbyid来模拟复杂页面布局初始化的耗时
 textview = (textview)findviewbyid(r.id.textview);
 }

 //展示预加载的数据
 subscriber<string> observer = new subscriber<string>() {
 @override public void oncompleted() { }

 @override public void onerror(throwable e) {
  e.printstacktrace();
 }

 @override
 public void onnext(string s) {
  textview.settext(s);
 }
 };

 private void preload() {
 preloader = rxpreloader.preload(observable.just("result").delay(500, timeunit.milliseconds));
 }

 @override
 protected void ondestroy() {
 super.ondestroy();
 preloader.destroy();//销毁
 }
}

最后，附上源码：http://xiazai.jb51.net/201701/yuanma/rxpreloader(jb51.net).rar

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。