Rxjava功能操作符的使用方法详解

compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
// Because RxAndroid releases are few and far between, it is recommended you also
// explicitly depend on RxJava's latest version for bug fixes and new features.
  compile 'io.reactivex.rxjava2:rxjava:2.1.5'

compile 'com.alibaba:fastjson:1.2.39'

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.view.View;
import android.widget.TextView;
 
import com.alibaba.fastjson.JSONObject;
 
import java.io.IOException;
import java.util.concurrent.TimeUnit;
 
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;
import okhttp3.Call;
import okhttp3.Callback;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
 
public class MainActivity extends AppCompatActivity {
 
  private TextView name;
 
  @Override
  protected void onCreate(Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    setContentView(R.layout.activity_main);
 
    name = (TextView) findViewById(R.id.name);
    //用来调用下面的方法，监听。
    name.setOnClickListener(new View.OnClickListener() {
      @Override
      public void onClick(View v) {
 
        interval();
      }
    });
  }
 
  //例1：Observer
  public void observer() {
    //观察者
    Observer<string> observer = new Observer<string>() {
      @Override
      public void onSubscribe(@NonNull Disposable d) {
 
      }
      @Override
      public void onNext(@NonNull String s) {
        //接收从被观察者中返回的数据
        System.out.println("onNext :" + s);
      }
      @Override
      public void onError(@NonNull Throwable e) {
 
      }
      @Override
      public void onComplete() {
 
      }
    };
    //被观察者
    Observable<string> observable = new Observable<string>() {
      @Override
      protected void subscribeActual(Observer<!--? super String--> observer) {
        observer.onNext("11111");
        observer.onNext("22222");
        observer.onComplete();
      }
    };
    //产生了订阅
    observable.subscribe(observer);
  }
 
  //例2：Flowable
  private void flowable(){
    //被观察者
    Flowable.create(new FlowableOnSubscribe<string>() {
      @Override
      public void subscribe(@NonNull FlowableEmitter<string> e) throws Exception {
        for (int i = 0; i < 100; i++) {
          e.onNext(i+"");
        }
      }
      //背压的策略，buffer缓冲区        观察者
      //背压一共给了五种策略
      // BUFFER、
      // DROP、打印前128个，后面的删除
      // ERROR、
      // LATEST、打印前128个和最后一个，其余删除
      // MISSING
      //这里的策略若不是BUFFER 那么，会出现著名的：MissingBackpressureException错误
    }, BackpressureStrategy.BUFFER).subscribe(new Consumer<string>() {
      @Override
      public void accept(String s) throws Exception {
        System.out.println("subscribe accept"+s);
        Thread.sleep(1000);
      }
    });
  }
 
  //例3：线程调度器 Scheduler
  public void flowable1(){
    Flowable.create(new FlowableOnSubscribe<string>() {
      @Override
      public void subscribe(@NonNull FlowableEmitter<string> e) throws Exception {
        for (int i = 0; i < 100; i++) {
          //输出在哪个线程
          System.out.println("subscribe Thread.currentThread.getName = " + Thread.currentThread().getName());
          e.onNext(i+"");
        }
      }
    },BackpressureStrategy.BUFFER)
        //被观察者一般放在子线程
        .subscribeOn(Schedulers.io())
        //观察者一般放在主线程
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<string>() {
          @Override
          public void accept(String s) throws Exception {
            System.out.println("s"+ s);
            Thread.sleep(100);
            //输出在哪个线程
            System.out.println("subscribe Thread.currentThread.getName = " + Thread.currentThread().getName());
          }
        });
  }
 
 
  //例4：http请求网络，map转化器，fastjson解析器
  public void map1(){
    Observable.create(new ObservableOnSubscribe<string>() {
      @Override
      public void subscribe(@NonNull final ObservableEmitter<string> e) throws Exception {
        OkHttpClient client = new OkHttpClient();
        Request request = new Request.Builder()
            .url("https://qhb.2dyt.com/Bwei/login")
            .build();
        client.newCall(request).enqueue(new Callback() {
          @Override
          public void onFailure(Call call, IOException e) {
 
          }
 
          @Override
          public void onResponse(Call call, Response response) throws IOException {
            String result = response.body().string();
            e.onNext(result);
          }
        });
      }
    })
        //map转换器 flatmap(无序)，concatmap(有序)
        .map(new Function<string, bean="">() {
      @Override
      public Bean apply(@NonNull String s) throws Exception {
        //用fastjson来解析数据
        return JSONObject.parseObject(s,Bean.class);
      }
    }).subscribe(new Consumer<bean>() {
      @Override
      public void accept(Bean bean) throws Exception {
        System.out.println("bean = "+ bean.toString() );
      }
    });
  }
 
  //常见rxjava操作符
  //例 定时发送消息
  public void interval(){
    Observable.interval(2,1, TimeUnit.SECONDS)
        .take(10)
        .subscribe(new Consumer<long>() {
          @Override
          public void accept(Long aLong) throws Exception {
            System.out.println("aLong = " + aLong);
          }
        });
  }
 
 
  //例 zip字符串合并
  public void zip(){
    Observable observable1 = Observable.create(new ObservableOnSubscribe<string>() {
      @Override
      public void subscribe(@NonNull ObservableEmitter<string> e) throws Exception {
        e.onNext("1");
        e.onNext("2");
        e.onNext("3");
        e.onNext("4");
        e.onComplete();
 
      }
    });
    Observable observable2 = Observable.create(new ObservableOnSubscribe<string>() {
      @Override
      public void subscribe(@NonNull ObservableEmitter<string> e) throws Exception {
        e.onNext("A");
        e.onNext("B");
        e.onNext("C");
        e.onNext("D");
        e.onComplete();
      }
    });
 
    Observable.zip(observable1, observable2, new BiFunction<string,string,string>() {
      @Override
      public String apply(@NonNull String o, @NonNull String o2) throws Exception {
        return o + o2;
      }
    }).subscribe(new Consumer<string>() {
      @Override
      public void accept(String o) throws Exception {
        System.out.println("o"+ o);
      }
    });
  }