Rxjava响应式编程

一、Rxjava的思维

（1）响应式编程 （根据上一层的响应来影响下一层的变化） Rx全称:reactivex

链式编程：起点-需求1-需求2-......-终点

eg：登录操作： 触发登录按钮（起点）------第一步登录--------第二步登录------第三步登录--------登录跳转至主页面（终点）

中间三步登录即为需求

（2）具体思维代码:

只看起点和终点：obsevrable被观察者为起点、subscribe订阅为关联起点和终点

1.  
   Observable.just(Path)
2.  
         //关联：观察者设计模式 关联起点和终点=订阅
3.  
         .subscribe(new Observer<Bitmap>() {
4.  
             //订阅成功
5.  
             @Override
6.  
             public void onSubscribe(@io.reactivex.rxjava3.annotations.NonNull Disposable d) {
7.  
                 pg.show("正在加载中"); //终点进程的第一步
8.  
9.  
             }
10.  
              //上一层给我的响应
11.  
              @Override
12.  
              public void onNext(@io.reactivex.rxjava3.annotations.NonNull Bitmap bitmap) {
13.  
                  binding.iv.setImageBitmap(bitmap); //响应后显示图片
14.  
15.  
              }
16.  
              //链条思维发生异常
17.  
18.  
              @Override
19.  
              public void onError(@io.reactivex.rxjava3.annotations.NonNull Throwable e) {
20.  
21.  
              }
22.  
              //整个链条全部结束
23.  
              @Override
24.  
              public void onComplete() {
25.  
                  if(pg!=null){
26.  
                      pg.dismiss(); //隐藏进度条
27.  
                  }
28.  
29.  
              }
30.  
          });

如果中间添加了三个需求：

1.  
   Observable.just(Path)
2.  
         //需求1：图片下载需求将string转成bitmap
3.  
         .map(new Function<String, Bitmap>() {
4.  
             @Override
5.  
             public Bitmap apply(String path) throws Throwable {
6.  
                 try{
7.  
                     Thread.sleep(2000);
8.  
                     //请求服务器响应数据
9.  
                     URL url=new URL(path);
10.  
                      HttpURLConnection httpURLConnection= (HttpURLConnection) url.openConnection();
11.  
                      httpURLConnection.setConnectTimeout(5000); //设置请求最短时间
12.  
                      int responseCode=httpURLConnection.getResponseCode(); //得到响应码
13.  
                      if(responseCode==HttpURLConnection.HTTP_OK){
14.  
                          InputStream inputStream=httpURLConnection.getInputStream();
15.  
                        Bitmap bitmap=BitmapFactory.decodeStream(inputStream);
16.  
                        return bitmap; //将此时的bitmap传至下一层
17.  
18.  
                      }
19.  
20.  
                  }catch (Exception e){
21.  
                      e.printStackTrace();
22.  
                  }
23.  
                  return null;
24.  
              }
25.  
          })
26.  
27.  
          //需求2：加水印
28.  
          .map(new Function<Bitmap, Bitmap>() {
29.  
              @Override
30.  
              public Bitmap apply(Bitmap bitmap) throws Throwable {
31.  
                  Paint paint=new Paint(); //画布
32.  
                  paint.setColor(Color.BLUE);
33.  
                  paint.setTextSize(88);   // 字的大小
34.  
                  Bitmap shuiyingbitmap=drawTextToBitmap(bitmap,"好的",paint,88,88);
35.  
                  return shuiyingbitmap; //将此时加水印的bitmap传至下一层
36.  
              }
37.  
          })
38.  
           
39.  
          //需求3：日志打印需求
40.  
          .map(new Function<Bitmap, Bitmap>() {
41.  
              @Override
42.  
              public Bitmap apply(Bitmap bitmap) throws Throwable {
43.  
                  Log.e("tag","什么时候下载了图片apply" System.currentTimeMillis());
44.  
                  return bitmap;   //此时的bitmap没有做任何改变
45.  
              }
46.  
          })
47.  
           
48.  
          //给上面的线程分配异步线程
49.  
          .subscribeOn(Schedulers.io())
50.  
           
51.  
          //终点分配Android主线程
52.  
          .observeOn(AndroidSchedulers.mainThread())
53.  
           
54.  
          //关联：观察者设计模式 关联起点和终点=订阅
55.  
          .subscribe(new Observer<Bitmap>() {
56.  
              //订阅成功
57.  
              @Override
58.  
              public void onSubscribe(@io.reactivex.rxjava3.annotations.NonNull Disposable d) {
59.  
                  pg.show("正在加载中"); //终点进程的第一步
60.  
61.  
              }
62.  
              //上一层给我的响应
63.  
              @Override
64.  
              public void onNext(@io.reactivex.rxjava3.annotations.NonNull Bitmap bitmap) {
65.  
                  binding.iv.setImageBitmap(bitmap); //响应后显示图片
66.  
67.  
              }
68.  
              //链条思维发生异常
69.  
70.  
              @Override
71.  
              public void onError(@io.reactivex.rxjava3.annotations.NonNull Throwable e) {
72.  
73.  
              }
74.  
              //整个链条全部结束
75.  
              @Override
76.  
              public void onComplete() {
77.  
                  if(pg!=null){
78.  
                      pg.dismiss(); //隐藏进度条
79.  
                  }
80.  
81.  
              }
82.  
          });

二、订阅及主要的操作符

（1）subscribe中observe和custom的区别：

observe中有四个方法：onSubscribe、onNext、 onError、onComplete()

custom中只有一个方法：accept（）对应的observe中的onNext（）方法

1.  
   Observable.fromArray(strings)
2.  
         //订阅起点终点
3.  
   .subscribe(new Consumer<String>() {
4.  
     @Override
5.  
     //终点
6.  
     public void accept(String s) throws Throwable {
7.  
         Log.d("tag","accept:" s); //打印输出数组
8.  
     }
9.  
   });

（2）主要的操作符

1.创建操作符 create、just 主要是创建observeable

2.转换操作符 map、flatmap 主要是对发射的事件进行再处理，比如上面的增加需求用map操作符

3.组合操作符：concat、concatArray 主要就是将多个被观察者集合成一个呗观察者

4.功能操作符：subscribeOn()：决定产生事件发射的线程、observeOn（）：决定下游事件被处理时所处的线程

这篇好文章是转载于：学新通技术网