Java中ReactiveX（RxJava）的使用

1.1 ReactiveX概述

ReactiveX官网：ReactiveX

1.1.1 Android的MVP开发模式

MVP的工作流程

• Presenter负责逻辑的处理
• Model提供数据
• View负责显示

作为一种新的模式，在MVP中View并不直接使用Model，它们之间的通信是通过Presenter来进行的，所有的交互都发生在Presenter内部，而在MVC中View会从直接Model中读取数据而不是通过 Controller。

1.1.2 ReactiveX简介

RX（ReactiveX）<==> 反应式 <==> 响应式编程思维

什么是响应式编程：根据上一层的响应来影响下一层的变化

1.1.3 为什么要学习ReactiveX

• 改变思维（响应式编程思维)，编程效率提升

1.2  RxJava使用

RxJava地址：ReactiveX/RxJava (github.com)

RxAndroid地址：ReactiveX/RxAndroid (github.com)

1.2.1 RxJava的使用步骤

1. 引入RxJava

   implementation 'io.reactivex.rxjava3:rxandroid:3.0.0'
   implementation 'io.reactivex.rxjava3:rxjava:3.1.4'

2. 创建 Observer(被观察者对象)

   //Observable部分,被观察者部分
   Observable myObservable=Observable.create(new Observable.OnSubscribe() {
       @Override
       public void call(Subscriber subscriber) {
           subscriber.onNext("我是被观察的对象");
           subscriber.onCompleted();
       }
   });

3. 创建Subscriber(理解为观察者对象)

   //Subscriber部分，观察者部分
   Subscriber mySubscriber=new Subscriber() {
       @Override
       public void onCompleted() {
   
       }
   
       @Override
       public void onError(Throwable e) {
   
       }
   
       @Override
       public void onNext(String s) {
           System.out.println("接收到‘"+s+"’");;
       }
   };

4. Observer和Subscriber关联

    myObservable.subscribe(mySubscriber);

1.2.2 使用RxJava加载图片

activity_main.xml代码

MainActivity.java代码

package com.example.myrxjava;

import androidx.appcompat.app.AppCompatActivity;

import android.app.ProgressDialog;
import android.graphics.Bitmap;
import android.graphics.BitmapFactory;
import android.os.Bundle;
import android.view.View;
import android.widget.Button;
import android.widget.ImageView;

import java.io.InputStream;
import java.net.HttpURLConnection;
import java.net.URL;

import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class MainActivity extends AppCompatActivity implements View.OnClickListener {

    private Button btn_showImage;
    private ImageView iv_image;

    private final String PATH = "http://www.kaotop.com/file/tupian/20220505/apic2597.jpg";
    private ProgressDialog progressDialog;

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

        btn_showImage = findViewById(R.id.btn_showImage);
        iv_image = findViewById(R.id.iv_image);

        btn_showImage.setOnClickListener(this);
    }

    @Override
    public void onClick(View view) {
        switch (view.getId()){
            case R.id.btn_showImage:
                showImageAction(view);
                break;
        }
    }

    public void showImageAction(View view){
        //RX系列框架为什么把所有函数都称为 *** 作符：因为我们的函数要去 *** 作从起点流向终点的过程
        //观察者模式：被观察者Observable ==> 观察者Observer
        //起点
        Observable.just(PATH) //第二步，分发将请求地址的字符串
                .map(new Function() { //第三步，转换为图片给下一层
                    @Override
                    public Bitmap apply(String path) throws Throwable {
                        Thread.sleep(3000); //模拟网络延迟
                        URL url = new URL(path);
                        HttpURLConnection httpURLConnection = (HttpURLConnection) url.openConnection();
                        httpURLConnection.setConnectTimeout(5000); //设置请求超时时长
                        int responseCode = httpURLConnection.getResponseCode();
                        if (responseCode == HttpURLConnection.HTTP_OK){
                            InputStream inputStream = httpURLConnection.getInputStream();
                            Bitmap bitmap = BitmapFactory.decodeStream(inputStream);
                            return bitmap;
                        }
                        return null;
                    }
                })
                .subscribeOn(Schedulers.io()) //改变调用它之前代码的线程。分配请求异步线程
                .observeOn(AndroidSchedulers.mainThread()) //改变调用它之后代码的线程。给终点分配主线程
                .subscribe(new Observer() { //订阅（关联）
                    @Override
                    public void onSubscribe(@NonNull Disposable d) { //第一步，订阅成功，显示加载框
                        progressDialog = new ProgressDialog(MainActivity.this);
                        progressDialog.setTitle("图片加载中...");
                        progressDialog.show();
                    }

                    @Override
                    public void onNext(@NonNull Bitmap bitmap) { //第四步，处理上一层返回的响应
                        iv_image.setImageBitmap(bitmap);
                    }

                    @Override
                    public void onError(@NonNull Throwable e) { //业务流程链发生异常
                        e.printStackTrace();
                    }

                    @Override
                    public void onComplete() { //第五步，业务流程链全部结束
                        if (progressDialog != null) {
                            progressDialog.dismiss();
                        }
                    }
                });
    }
}

运行效果图

1.2.3 在1.2.2的基础上添加 *** 作

drawTextToBitmap()是一个自定义的为图片添加水印的方法 

1.2.4 在1.2.3的基础上添加打印日志

总结：可以在起点和终点之间任意添加各种 *** 作/需求

1.2.5 使用Rx思维遍历数组

代码：

    private void action() {
        String[] strings = {"AAA", "BBB", "CCC", "DDD"};
        Observable.fromArray(strings)
                .subscribe(new Consumer() {
                    @Override
                    public void accept(String s) throws Throwable {
                        Log.e(TAG, "accept: " + s);;
                    }
                });
    }

运行效果：

1.2.6 使用Rx思维解决问题

问题描述：服务器成功返回给客服端的data数据一般是JavaBean数据。当服务器返回数据失败时，data数据就不一定是再是JavaBean，而是可能是null、0等不同的数据，故解析时就会出现异常，下面就使用Rx来解决此问题

步骤流程图： 

Step1：创建总Bean

    private SuccessBean data;
    private int code;
    private String message;

Step2：创建成功Bean（data类）

    private int id;
    private String name;

Step3：构建登录引擎起点

import com.example.myrxjava.login.bean.ResponseResult;
import com.example.myrxjava.login.bean.SuccessBean;

import io.reactivex.rxjava3.core.Observable;

public class LoginEngine {
    public static Observable login(String username, String password){
        ResponseResult responseResult = new ResponseResult();
        if ("zhumeng".equals(username) && "123456".equals(password)){
            SuccessBean successBean = new SuccessBean();
            successBean.setId(45678910);
            successBean.setName("zhumeng登录成功");

            responseResult.setData(successBean);
            responseResult.setCode(200);
            responseResult.setMessage("登录成功");
        } else {
            responseResult.setData(null);
            responseResult.setCode(404);
            responseResult.setMessage("登录失败");
        }
        return Observable.just(responseResult); //返回响应的起点
    }
}

Step4：创建中间处理对象Customobserver

import com.example.myrxjava.login.bean.ResponseResult;
import com.example.myrxjava.login.bean.SuccessBean;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;

public abstract class CustomObserver implements Observer {
    public abstract void success(SuccessBean successBean);
    public abstract void error(String messge);

    @Override
    public void onSubscribe(@NonNull Disposable d) {

    }

    @Override
    public void onNext(@NonNull ResponseResult responseResult) {
        if (responseResult.getCode() == 200){
            success(responseResult.getData());
        } else {
            error(responseResult.getMessage() + "，请求失败，请检查日志");
        }
    }

    @Override
    public void onError(@NonNull Throwable e) {
        error(e.getMessage() + "，请检查日志获取错误信息详情");
    }

    @Override
    public void onComplete() {

    }
}

Step5：实现Rx思维的登录 *** 作

    private void login(String username, String password) {
        LoginEngine.login(username, password)
                .subscribe(new CustomObserver() {
                    @Override
                    public void success(SuccessBean successBean) {
                        Log.d(TAG, "success: 成功Bean详情为" + successBean.toString());
                    }

                    @Override
                    public void error(String messge) {
                        Log.d(TAG, "error: 失败的Message信息为" + messge);
                    }
                });
    }

Step6：运行效果