一、概念

Retrofit之所以作为现在最流行的网络请求库，其中一个主要原因是支持RxJava，即Retrofit除了提供传统的网络请求方式外，还提供RxJava版本的网络请求方式。两种方式在使用上最大的区别在于：传统方式采用了Callback接口，而RxJava方式则采用了Observable接口。

主要体现在：
1.用于描述网络请求的接口的设置

// 传统方式：Call<..>接口形式
public interface GetRequest_Interface {
    @GET("url地址")
    Call<Translation> getCall();
    // 注解里传入 网络请求 的部分URL地址
    // getCall()是接受网络请求数据的方法
}


// RxJava方式：Observable<..>接口形式
@GET("url地址")
Observable<Translation> getCall();

2.网络请求的封装形式和发送形式

// 传统方式
// 1. 创建 网络请求接口 的实例
GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);

// 2. 采用Call<..>接口 对 发送请求 进行封装
Call<Translation> call = request.getCall();

// 3. 发送网络请求(异步)
call.enqueue(new Callback<Translation>() {
    // 请求成功时回调
    @Override
    public void onResponse(Call<Translation> call, Response<Translation> response) {
         ...  
    }

    // 请求失败时回调
    @Override
    public void onFailure(Call<Translation> call, Throwable throwable) {
        ....
    }
});


// RxJava方式
// 1. 创建 网络请求接口 的实例
GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);

// 2. 采用Observable<...>形式 对 网络请求 进行封装
Observable<Translation> observable = request.getCall();

// 3. 发送网络请求(异步)
observable.subscribeOn(Schedulers.io())               // 在IO线程进行网络请求
          .observeOn(AndroidSchedulers.mainThread())  // 回到主线程 处理请求结果
          .subscribe(new Observer<Translation>() {

            // 发送请求后调用该复写方法（无论请求成功与否）
            @Override
            public void onSubscribe(Disposable d) {
                ...// 初始化工作
              }
            
            // 发送请求成功后调用该复写方法
            @Override
            public void onNext(Translation result) {
                ...// 对返回结果Translation类对象 进行处理
            }

            // 发送请求成功后，先调用onNext（）再调用该复写方法
            @Override
            public void onComplete() {
                Log.d(TAG, "请求成功");
            }
            // 发送请求失败后调用该复写方法
            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "请求失败");
            }

        });
}

二、使用

1.基本使用

步骤1：添加依赖及网络权限

//添加依赖
// Android 支持 Rxjava
// 此处一定要注意使用RxJava2的版本
compile 'io.reactivex.rxjava2:rxjava:2.0.1'
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'

// Android 支持 Retrofit
compile 'com.squareup.retrofit2:retrofit:2.1.0'

// 衔接 Retrofit & RxJava
// 此处一定要注意使用RxJava2的版本
compile 'com.jakewharton.retrofit:retrofit2-rxjava2-adapter:1.0.0'

// 支持Gson解析
compile 'com.squareup.retrofit2:converter-gson:2.1.0'


//添加网络权限
<uses-permission android:name="android.permission.INTERNET"/>

步骤2：创建 接收服务器返回数据 的类

public class Translation {
    private int status;

    private content content;
    private static class content {
        private String from;
        private String to;
        private String vendor;
        private String out;
        private int errNo;
    }

    //定义 输出返回数据 的方法
    public void show() {
        System.out.println( "Rxjava翻译结果：" + status);
        System.out.println("Rxjava翻译结果：" + content.from);
        System.out.println("Rxjava翻译结果：" + content.to);
        System.out.println("Rxjava翻译结果：" + content.vendor);
        System.out.println("Rxjava翻译结果：" + content.out);
        System.out.println("Rxjava翻译结果：" + content.errNo);
    }
}

步骤3：创建 用于描述网络请求 的接口（区别于传统形式）
采用 注解 + Observable<...>接口描述 网络请求参数。

public interface GetRequest_Interface {

    @GET("ajax.php?a=fy&f=auto&t=auto&w=hi%20world")
    Observable<Translation> getCall();
     // 注解里传入 网络请求 的部分URL地址
    // Retrofit把网络请求的URL分成了两部分：一部分放在Retrofit对象里，另一部分放在网络请求接口里
    // 如果接口里的url是一个完整的网址，那么放在Retrofit对象里的URL可以忽略
    // 采用Observable<...>接口 
    // getCall()是接受网络请求数据的方法
}

步骤4：创建 Retrofit 实例

Retrofit retrofit = new Retrofit.Builder()
                .baseUrl("http://fy.iciba.com/") // 设置 网络请求 Url
                .addConverterFactory(GsonConverterFactory.create()) //设置使用Gson解析(记得加入依赖)
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // 支持RxJava
                .build();

步骤5：创建 网络请求接口 实例

GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);

步骤6：对 网络请求 进行封装
采用Observable<...>形式 。

Observable<Translation> observable = request.getCall();

步骤7：发送网络请求
步骤8：对返回的数据进行处理

// 步骤7：发送网络请求
observable.subscribeOn(Schedulers.io())               // 在IO线程进行网络请求
          .observeOn(AndroidSchedulers.mainThread())  // 回到主线程处理请求结果
          .subscribe(new Observer<Translation>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "开始采用subscribe连接");
            }

            @Override
            public void onNext(Translation result) {
                // 步骤8：对返回的数据进行处理
                result.show() ;
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "请求失败");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "请求成功");
            }
        });

2.无条件网络请求轮询

public class RxJavafixRxjava extends AppCompatActivity {

    private static final String TAG = "Rxjava";

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

        /*
         * 步骤1：采用interval（）延迟发送
         * 注：此处主要展示无限次轮询，若要实现有限次轮询，仅需将interval（）改成intervalRange（）即可
         **/
        Observable.interval(2,1, TimeUnit.SECONDS)
                // 参数说明：
                // 参数1 = 第1次延迟时间；
                // 参数2 = 间隔时间数字；
                // 参数3 = 时间单位；
                // 该例子发送的事件特点：延迟2s后发送事件，每隔1秒产生1个数字（从0开始递增1，无限个）

                 /*
                  * 步骤2：每次发送数字前发送1次网络请求（doOnNext（）在执行Next事件前调用）
                  * 即每隔1秒产生1个数字前，就发送1次网络请求，从而实现轮询需求
                  **/
                 .doOnNext(new Consumer<Long>() {
            @Override
            public void accept(Long integer) throws Exception {
                Log.d(TAG, "第 " + integer + " 次轮询" );

                 /*
                  * 步骤3：通过Retrofit发送网络请求
                  **/
                  // a. 创建Retrofit对象
                Retrofit retrofit = new Retrofit.Builder()
                        .baseUrl("http://fy.iciba.com/") // 设置 网络请求 Url
                        .addConverterFactory(GsonConverterFactory.create()) //设置使用Gson解析(记得加入依赖)
                        .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // 支持RxJava
                        .build();

                  // b. 创建 网络请求接口 的实例
                GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);

                  // c. 采用Observable<...>形式 对 网络请求 进行封装
                  Observable<Translation> observable = request.getCall();
                  // d. 通过线程切换发送网络请求
                  observable.subscribeOn(Schedulers.io())               // 切换到IO线程进行网络请求
                        .observeOn(AndroidSchedulers.mainThread())  // 切换回到主线程 处理请求结果
                        .subscribe(new Observer<Translation>() {
                            @Override
                            public void onSubscribe(Disposable d) {
                            }

                            @Override
                            public void onNext(Translation result) {
                                // e.接收服务器返回的数据
                                result.show() ;
                            }

                            @Override
                            public void onError(Throwable e) {
                                Log.d(TAG, "请求失败");
                            }

                            @Override
                            public void onComplete() {

                            }
                        });

            }
        }).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {

            }
            @Override
            public void onNext(Long value) {

            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "对Error事件作出响应");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "对Complete事件作出响应");
            }
        });


    }
}

3.有条件网络请求轮询

public class RxJavafixRetrofit extends AppCompatActivity {

    private static final String TAG = "Rxjava";

    // 设置变量 = 模拟轮询服务器次数
    private int i = 0 ;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        
        // 步骤1：创建Retrofit对象
        Retrofit retrofit = new Retrofit.Builder()
                .baseUrl("http://fy.iciba.com/") // 设置 网络请求 Url
                .addConverterFactory(GsonConverterFactory.create()) //设置使用Gson解析(记得加入依赖)
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // 支持RxJava
                .build();

        // 步骤2：创建 网络请求接口 的实例
        GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);

        // 步骤3：采用Observable<...>形式 对 网络请求 进行封装
        Observable<Translation> observable = request.getCall();

        // 步骤4：发送网络请求 & 通过repeatWhen（）进行轮询
        observable.repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
            @Override
            // 在Function函数中，必须对输入的 Observable<Object>进行处理，此处使用flatMap操作符接收上游的数据
            public ObservableSource<?> apply(@NonNull Observable<Object> objectObservable) throws Exception {
                // 将原始 Observable 停止发送事件的标识（Complete（） /  Error（））转换成1个 Object 类型数据传递给1个新被观察者（Observable）
                // 以此决定是否重新订阅 & 发送原来的 Observable，即轮询
                // 此处有2种情况：
                // 1. 若返回1个Complete（） /  Error（）事件，则不重新订阅 & 发送原来的 Observable，即轮询结束
                // 2. 若返回其余事件，则重新订阅 & 发送原来的 Observable，即继续轮询
                return objectObservable.flatMap(new Function<Object, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply(@NonNull Object throwable) throws Exception {

                        // 加入判断条件：当轮询次数 = 5次后，就停止轮询
                        if (i > 3) {
                            // 此处选择发送onError事件以结束轮询，因为可触发下游观察者的onError（）方法回调
                            return Observable.error(new Throwable("轮询结束"));
                        }
                        // 若轮询次数＜4次，则发送1Next事件以继续轮询
                        // 注：此处加入了delay操作符，作用 = 延迟一段时间发送（此处设置 = 2s），以实现轮询间间隔设置
                        return Observable.just(1).delay(2000, TimeUnit.MILLISECONDS);
                    }
                });

            }
        }).subscribeOn(Schedulers.io())               // 切换到IO线程进行网络请求
                .observeOn(AndroidSchedulers.mainThread())  // 切换回到主线程 处理请求结果
                .subscribe(new Observer<Translation>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                    }

                    @Override
                    public void onNext(Translation result) {
                        // e.接收服务器返回的数据
                        result.show() ;
                        i++;
                    }

                    @Override
                    public void onError(Throwable e) {
                        // 获取轮询结束信息
                        Log.d(TAG,  e.toString());
                    }

                    @Override
                    public void onComplete() {

                    }
                });

    }
}

4.网络请求嵌套回调

未用操作符，代码如下：

// 发送注册网络请求的函数方法
private void register() {
    api.register(new RegisterRequest())
            .subscribeOn(Schedulers.io())               //在IO线程进行网络请求
            .observeOn(AndroidSchedulers.mainThread())  //回到主线程去处理请求结果
            .subscribe(new Consumer<RegisterResponse>() {
                @Override
                public void accept(RegisterResponse registerResponse) throws Exception {
                    Toast.makeText(MainActivity.this, "注册成功", Toast.LENGTH_SHORT).show();
                    login();   //注册成功, 调用登录的方法
                }
            }, new Consumer<Throwable>() {
                @Override
                public void accept(Throwable throwable) throws Exception {
                    Toast.makeText(MainActivity.this, "注册失败", Toast.LENGTH_SHORT).show();
                }
            });
}


// 发送登录网络请求的函数方法
private void login() {
    api.login(new LoginRequest())
            .subscribeOn(Schedulers.io())               //在IO线程进行网络请求
            .observeOn(AndroidSchedulers.mainThread())  //回到主线程去处理请求结果
            .subscribe(new Consumer<LoginResponse>() {
                @Override
                public void accept(LoginResponse loginResponse) throws Exception {
                    Toast.makeText(MainActivity.this, "登录成功", Toast.LENGTH_SHORT).show();
                }
            }, new Consumer<Throwable>() {
                @Override
                public void accept(Throwable throwable) throws Exception {
                    Toast.makeText(MainActivity.this, "登录失败", Toast.LENGTH_SHORT).show();
                }
            });
} 

使用flatMap操作符，代码如下：

public class MainActivity extends AppCompatActivity {

        private static final String TAG = "Rxjava";

        // 定义Observable接口类型的网络请求对象
        Observable<Translation1> observable1;
        Observable<Translation2> observable2;

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

            // 步骤1：创建Retrofit对象
            Retrofit retrofit = new Retrofit.Builder()
                    .baseUrl("http://fy.iciba.com/") // 设置 网络请求 Url
                    .addConverterFactory(GsonConverterFactory.create()) //设置使用Gson解析(记得加入依赖)
                    .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // 支持RxJava
                    .build();

            // 步骤2：创建 网络请求接口 的实例
            GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);

            // 步骤3：采用Observable<...>形式 对 2个网络请求 进行封装
            observable1 = request.getCall();
            observable2 = request.getCall_2();


            observable1.subscribeOn(Schedulers.io())               // （初始被观察者）切换到IO线程进行网络请求1
                       .observeOn(AndroidSchedulers.mainThread())  // （新观察者）切换到主线程 处理网络请求1的结果
                       .doOnNext(new Consumer<Translation1>() {
                        @Override
                        public void accept(Translation1 result) throws Exception {
                            Log.d(TAG, "第1次网络请求成功");
                            result.show();
                            // 对第1次网络请求返回的结果进行操作 = 显示翻译结果
                        }
                    })

                    .observeOn(Schedulers.io())                 // （新被观察者，同时也是新观察者）切换到IO线程去发起登录请求
                                                                // 特别注意：因为flatMap是对初始被观察者作变换，所以对于旧被观察者，它是新观察者，所以通过observeOn切换线程
                                                                // 但对于初始观察者，它则是新的被观察者
                    .flatMap(new Function<Translation1, ObservableSource<Translation2>>() { // 作变换，即作嵌套网络请求
                        @Override
                        public ObservableSource<Translation2> apply(Translation1 result) throws Exception {
                            // 将网络请求1转换成网络请求2，即发送网络请求2
                            return observable2;
                        }
                    })

                    .observeOn(AndroidSchedulers.mainThread())  // （初始观察者）切换到主线程 处理网络请求2的结果
                    .subscribe(new Consumer<Translation2>() {
                        @Override
                        public void accept(Translation2 result) throws Exception {
                            Log.d(TAG, "第2次网络请求成功");
                            result.show();
                            // 对第2次网络请求返回的结果进行操作 = 显示翻译结果
                        }
                    }, new Consumer<Throwable>() {
                        @Override
                        public void accept(Throwable throwable) throws Exception {
                            System.out.println("登录失败");
                        }
                    });
    }
}

5.合并数据源 & 同时展示

使用merge操作符：

// 用于存放最终展示的数据
String result = "数据源来自 = " ;

/*
 * 设置第1个Observable：通过网络获取数据
 * 此处仅作网络请求的模拟
 **/
Observable<String> network = Observable.just("网络");

/*
 * 设置第2个Observable：通过本地文件获取数据
 * 此处仅作本地文件请求的模拟
 **/
Observable<String> file = Observable.just("本地文件");

/*
 * 通过merge（）合并事件 & 同时发送事件
 **/
Observable.merge(network, file)
        .subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String value) {
                Log.d(TAG, "数据源有： "+ value  );
                result += value + "+";
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "对Error事件作出响应");
            }

            // 接收合并事件后，统一展示
            @Override
            public void onComplete() {
                Log.d(TAG, "获取数据完成");
                Log.d(TAG,  result  );
            }
        });

使用zip操作符：

public class MainActivity extends AppCompatActivity {

    private static final String TAG = "Rxjava";

    // 定义Observable接口类型的网络请求对象
    Observable<Translation1> observable1;
    Observable<Translation2> observable2;

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

        // 步骤1：创建Retrofit对象
        Retrofit retrofit = new Retrofit.Builder()
                .baseUrl("http://fy.iciba.com/") // 设置 网络请求 Url
                .addConverterFactory(GsonConverterFactory.create()) //设置使用Gson解析(记得加入依赖)
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // 支持RxJava
                .build();

        // 步骤2：创建 网络请求接口 的实例
        GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);

        // 步骤3：采用Observable<...>形式 对 2个网络请求 进行封装
        observable1 = request.getCall().subscribeOn(Schedulers.io()); // 新开线程进行网络请求1
        observable2 = request.getCall_2().subscribeOn(Schedulers.io());// 新开线程进行网络请求2
        // 即2个网络请求异步 & 同时发送

        // 步骤4：通过使用zip（）对两个网络请求进行合并再发送
        Observable.zip(observable1, observable2,
                new BiFunction<Translation1, Translation2, String>() {
                    // 注：创建BiFunction对象传入的第3个参数 = 合并后数据的数据类型
                    @Override
                    public String apply(Translation1 translation1,
                                        Translation2 translation2) throws Exception {
                        return translation1.show() + " & " +translation2.show();
                    }
                }).observeOn(AndroidSchedulers.mainThread()) // 在主线程接收 & 处理数据
                .subscribe(new Consumer<String>() {
                    // 成功返回数据时调用
                    @Override
                    public void accept(String combine_infro) throws Exception {
                        // 结合显示2个网络请求的数据结果
                        Log.d(TAG, "最终接收到的数据是：" + combine_infro);
                    }
                }, new Consumer<Throwable>() {
                    // 网络请求错误时调用
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        System.out.println("登录失败");
                    }
                });
    }
}

6.从磁盘 / 内存缓存中获取缓存数据

// 该2变量用于模拟内存缓存 & 磁盘缓存中的数据
String memoryCache = null;
String diskCache = "从磁盘缓存中获取数据";

/*
 * 设置第1个Observable：检查内存缓存是否有该数据的缓存
 **/
Observable<String> memory = Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> emitter) throws Exception {

        // 先判断内存缓存有无数据
        if (memoryCache != null) {
            // 若有该数据，则发送
            emitter.onNext(memoryCache);
        } else {
            // 若无该数据，则直接发送结束事件
            emitter.onComplete();
        }

    }
});

/*
 * 设置第2个Observable：检查磁盘缓存是否有该数据的缓存
 **/
Observable<String> disk = Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> emitter) throws Exception {

        // 先判断磁盘缓存有无数据
        if (diskCache != null) {
            // 若有该数据，则发送
            emitter.onNext(diskCache);
        } else {
            // 若无该数据，则直接发送结束事件
            emitter.onComplete();
        }

    }
});

/*
 * 设置第3个Observable：通过网络获取数据
 **/
Observable<String> network = Observable.just("从网络中获取数据");
// 此处仅作网络请求的模拟

/*
 * 通过concat（） 和 firstElement（）操作符实现缓存功能
 **/

// 1. 通过concat（）合并memory、disk、network 3个被观察者的事件（即检查内存缓存、磁盘缓存 & 发送网络请求）
//    并将它们按顺序串联成队列
Observable.concat(memory, disk, network)
        // 2. 通过firstElement()，从串联队列中取出并发送第1个有效事件（Next事件），即依次判断检查memory、disk、network
        .firstElement()
        // 即本例的逻辑为：
        // a. firstElement()取出第1个事件 = memory，即先判断内存缓存中有无数据缓存；由于memoryCache = null，即内存缓存中无数据，所以发送结束事件（视为无效事件）
        // b. firstElement()继续取出第2个事件 = disk，即判断磁盘缓存中有无数据缓存：由于diskCache ≠ null，即磁盘缓存中有数据，所以发送Next事件（有效事件）
        // c. 即firstElement()已发出第1个有效事件（disk事件），所以停止判断。
        
        // 3. 观察者订阅
        .subscribe(new Consumer<String>() {
            @Override
            public void accept( String s) throws Exception {
                Log.d(TAG,"最终获取的数据来源 =  "+ s);
            }
        });

7.联合判断多个事件

/*
 * 步骤1：设置控件变量 & 绑定
 **/
EditText name,age,job;
Button list;

name = (EditText) findViewById(R.id.name);
age = (EditText) findViewById(R.id.age);
job = (EditText) findViewById(R.id.job);
list = (Button) findViewById(R.id.list);

/*
 * 步骤2：为每个EditText设置被观察者，用于发送监听事件
 * 说明：
 * 1. 此处采用了RxBinding：RxTextView.textChanges(name) = 对对控件数据变更进行监听（功能类似TextWatcher），需要引入依赖：compile 'com.jakewharton.rxbinding2:rxbinding:2.0.0'
 * 2. 传入EditText控件，点击任1个EditText撰写时，都会发送数据事件 = Function3（）的返回值（下面会详细说明）
 * 3. 采用skip(1)原因：跳过 一开始EditText无任何输入时的空值
 **/
Observable<CharSequence> nameObservable = RxTextView.textChanges(name).skip(1);
Observable<CharSequence> ageObservable = RxTextView.textChanges(age).skip(1);
Observable<CharSequence> jobObservable = RxTextView.textChanges(job).skip(1);

/*
 * 步骤3：通过combineLatest（）合并事件 & 联合判断
 **/
Observable.combineLatest(nameObservable,ageObservable,jobObservable,new Function3<CharSequence, CharSequence, CharSequence,Boolean>() {
    @Override
    public Boolean apply(@NonNull CharSequence charSequence, @NonNull CharSequence charSequence2, @NonNull CharSequence charSequence3) throws Exception {

        /*
         * 步骤4：规定表单信息输入不能为空
         **/
        // 1. 姓名信息
        boolean isUserNameValid = !TextUtils.isEmpty(name.getText()) ;
        // 除了设置为空，也可设置长度限制
        // boolean isUserNameValid = !TextUtils.isEmpty(name.getText()) && (name.getText().toString().length() > 2 && name.getText().toString().length() < 9);

        // 2. 年龄信息
        boolean isUserAgeValid = !TextUtils.isEmpty(age.getText());
        // 3. 职业信息
        boolean isUserJobValid = !TextUtils.isEmpty(job.getText()) ;

        /*
         * 步骤5：返回信息 = 联合判断，即3个信息同时已填写，"提交按钮"才可点击
         **/
        return isUserNameValid && isUserAgeValid && isUserJobValid;
    }

        }).subscribe(new Consumer<Boolean>() {
    @Override
    public void accept(Boolean s) throws Exception {
        /*
         * 步骤6：返回结果 & 设置按钮可点击样式
         **/
        Log.e(TAG, "提交按钮是否可点击： "+s);
        list.setEnabled(s);
    }
});

8.线程控制（切换 / 调度）

public class MainActivity extends AppCompatActivity {

    private static final String TAG = "Rxjava";

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

        //步骤4：创建Retrofit对象
        Retrofit retrofit = new Retrofit.Builder()
                .baseUrl("http://fy.iciba.com/") // 设置 网络请求 Url
                .addConverterFactory(GsonConverterFactory.create()) //设置使用Gson解析(记得加入依赖)
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // 支持RxJava
                .build();

        // 步骤5：创建 网络请求接口 的实例
        GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);

        // 步骤6：采用Observable<...>形式 对 网络请求 进行封装
        Observable<Translation> observable = request.getCall();

        // 步骤7：发送网络请求
        observable.subscribeOn(Schedulers.io())               // 在IO线程进行网络请求
                  .observeOn(AndroidSchedulers.mainThread())  // 回到主线程 处理请求结果
                  .subscribe(new Observer<Translation>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(TAG, "开始采用subscribe连接");
                    }

                    @Override
                    public void onNext(Translation result) {
                        // 步骤8：对返回的数据进行处理
                        result.show() ;
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(TAG, "请求失败");
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "请求成功");
                    }
                });
    }
}

9.网络请求出错重连

public class RxJavafixRetrofit2 extends AppCompatActivity {

    private static final String TAG = "RxJava";

    // 设置变量
    // 可重试次数
    private int maxConnectCount = 10;
    // 当前已重试次数
    private int currentRetryCount = 0;
    // 重试等待时间
    private int waitRetryTime = 0;

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

        // 步骤1：创建Retrofit对象
        Retrofit retrofit = new Retrofit.Builder()
                .baseUrl("http://fy.iciba.com/") // 设置 网络请求 Url
                .addConverterFactory(GsonConverterFactory.create()) //设置使用Gson解析(记得加入依赖)
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // 支持RxJava
                .build();

        // 步骤2：创建 网络请求接口 的实例
        GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);

        // 步骤3：采用Observable<...>形式 对 网络请求 进行封装
        Observable<Translation> observable = request.getCall();

        // 步骤4：发送网络请求 & 通过retryWhen（）进行重试
        // 注：主要异常才会回调retryWhen（）进行重试
        observable.retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(@NonNull Observable<Throwable> throwableObservable) throws Exception {
                // 参数Observable<Throwable>中的泛型 = 上游操作符抛出的异常，可通过该条件来判断异常的类型
                return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply(@NonNull Throwable throwable) throws Exception {

                        // 输出异常信息
                        Log.d(TAG,  "发生异常 = "+ throwable.toString());

                        /**
                         * 需求1：根据异常类型选择是否重试
                         * 即，当发生的异常 = 网络异常 = IO异常 才选择重试
                         */
                        if (throwable instanceof IOException){

                            Log.d(TAG,  "属于IO异常，需重试" );

                            /**
                             * 需求2：限制重试次数
                             * 即，当已重试次数 < 设置的重试次数，才选择重试
                             */
                            if (currentRetryCount < maxConnectCount){

                                // 记录重试次数
                                currentRetryCount++;
                                Log.d(TAG,  "重试次数 = " + currentRetryCount);

                                /**
                                 * 需求2：实现重试
                                 * 通过返回的Observable发送的事件 = Next事件，从而使得retryWhen（）重订阅，最终实现重试功能
                                 *
                                 * 需求3：延迟1段时间再重试
                                 * 采用delay操作符 = 延迟一段时间发送，以实现重试间隔设置
                                 *
                                 * 需求4：遇到的异常越多，时间越长
                                 * 在delay操作符的等待时间内设置 = 每重试1次，增多延迟重试时间1s
                                 */
                                // 设置等待时间
                                waitRetryTime = 1000 + currentRetryCount* 1000;
                                Log.d(TAG,  "等待时间 =" + waitRetryTime);
                                return Observable.just(1).delay(waitRetryTime, TimeUnit.MILLISECONDS);


                            }else{
                                // 若重试次数已 > 设置重试次数，则不重试
                                // 通过发送error来停止重试（可在观察者的onError（）中获取信息）
                                return Observable.error(new Throwable("重试次数已超过设置次数 = " +currentRetryCount  + "，即 不再重试"));

                            }
                        }

                        // 若发生的异常不属于I/O异常，则不重试
                        // 通过返回的Observable发送的事件 = Error事件 实现（可在观察者的onError（）中获取信息）
                        else{
                            return Observable.error(new Throwable("发生了非网络异常（非I/O异常）"));
                        }
                    }
                });
            }
        }).subscribeOn(Schedulers.io())               // 切换到IO线程进行网络请求
                .observeOn(AndroidSchedulers.mainThread())  // 切换回到主线程 处理请求结果
                .subscribe(new Observer<Translation>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                    }

                    @Override
                    public void onNext(Translation result) {
                        // 接收服务器返回的数据
                        Log.d(TAG,  "发送成功");
                        result.show();
                    }

                    @Override
                    public void onError(Throwable e) {
                        // 获取停止重试的信息
                        Log.d(TAG,  e.toString());
                    }

                    @Override
                    public void onComplete() {

                    }
                });

    }
} 

三、例子

//build.gradle
implementation 'com.squareup.retrofit2:retrofit:2.0.2'
implementation 'com.squareup.retrofit2:converter-gson:2.4.0'
implementation 'com.jakewharton.retrofit:retrofit2-rxjava2-adapter:1.0.0'
implementation 'io.reactivex.rxjava2:rxjava:2.0.1'
implementation 'io.reactivex.rxjava2:rxandroid:2.0.1'

//AndroidManifest.xml
<uses-permission android:name="android.permission.INTERNET"/>

//Gist
public class Gist {
    Map<String, GistFile> files;

    @Override
    public String toString() {
        StringBuilder stringBuilder = new StringBuilder("[Gist]: ");
        if(files != null && files.size() > 0) {
            for(Map.Entry entry : files.entrySet()) {
                stringBuilder.append(entry.getKey())
                        .append(":")
                        .append(entry.getValue())
                        .append(" ");
            }
        } else {
            stringBuilder.append("null");
        }
        return stringBuilder.toString();
    }

    private static class GistFile {
        String content;

        @Override
        public String toString() {
            return content;
        }
    }
}

//HttpService
public interface HttpService {

    @GET("gists/c2a7c39532239ff261be")
    Observable<Gist> getInfo();
}

//MainActivity
public class MainActivity extends AppCompatActivity {
    private static final String TAG = "Retrofit";

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        Log.d(TAG, "zwm, onCreate");
        Retrofit retrofit = new Retrofit.Builder()
                .baseUrl("https://api.github.com/")
                .addConverterFactory(GsonConverterFactory.create())
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .build();
        HttpService httpService = retrofit.create(HttpService.class);
        Observable<Gist> observable = httpService.getInfo();
        observable.subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Gist>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(TAG, "zwm, onSubscribe, thread: " + Thread.currentThread().getName());
                    }

                    @Override
                    public void onNext(Gist value) {
                        Log.d(TAG, "zwm, onNext, thread: " + Thread.currentThread().getName());
                        Log.d(TAG, "zwm, onNext: " + value);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(TAG, "zwm, onError, thread: " + Thread.currentThread().getName());
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "zwm, onComplete, thread: " + Thread.currentThread().getName());
                    }
                });
    }
}

//输出log
2019-12-31 10:20:38.748 7719-7719/com.example.sourcecodetest D/Retrofit: zwm, onCreate
2019-12-31 10:20:38.963 7719-7719/com.example.sourcecodetest D/Retrofit: zwm, onSubscribe, thread: main
2019-12-31 10:20:41.654 7719-7719/com.example.sourcecodetest D/Retrofit: zwm, onNext, thread: main
2019-12-31 10:20:41.656 7719-7719/com.example.sourcecodetest D/Retrofit: zwm, onNext: [Gist]: OkHttp.txt:
                             \\           //
                              \\  .ooo.  //
                               .@@@@@@@@@.
                             :@@@@@@@@@@@@@:
                            :@@. '@@@@@' .@@:
                            @@@@@@@@@@@@@@@@@
                            @@@@@@@@@@@@@@@@@
    
                       :@@ :@@@@@@@@@@@@@@@@@. @@:
                       @@@ '@@@@@@@@@@@@@@@@@, @@@
                       @@@ '@@@@@@@@@@@@@@@@@, @@@
                       @@@ '@@@@@@@@@@@@@@@@@, @@@
                       @@@ '@@@@@@@@@@@@@@@@@, @@@
                       @@@ '@@@@@@@@@@@@@@@@@, @@@
                       @@@ '@@@@@@@@@@@@@@@@@, @@@
                            @@@@@@@@@@@@@@@@@
                            '@@@@@@@@@@@@@@@'
                               @@@@   @@@@
                               @@@@   @@@@
                               @@@@   @@@@
                               '@@'   '@@'
    
         :@@@.
       .@@@@@@@:   +@@       `@@      @@`   @@     @@
      .@@@@'@@@@:  +@@       `@@      @@`   @@     @@
      @@@     @@@  +@@       `@@      @@`   @@     @@
     .@@       @@: +@@   @@@ `@@      @@` @@@@@@ @@@@@@  @@;@@@@@
     @@@       @@@ +@@  @@@  `@@      @@` @@@@@@ @@@@@@  @@@@@@@@@
     @@@       @@@ +@@ @@@   `@@@@@@@@@@`   @@     @@    @@@   :@@
     @@@       @@@ +@@@@@    `@@@@@@@@@@`   @@     @@    @@#    @@+
     @@@       @@@ +@@@@@+   `@@      @@`   @@     @@    @@:    @@#
      @@:     .@@` +@@@+@@   `@@      @@`   @@     @@    @@#    @@+
      @@@.   .@@@  +@@  @@@  `@@      @@`   @@     @@    @@@   ,@@
       @@@@@@@@@   +@@   @@@ `@@      @@`   @@@@   @@@@  @@@@#@@@@
        @@@@@@@    +@@   #@@ `@@      @@`   @@@@:  @@@@: @@'@@@@@
                                                         @@:
                                                         @@:
                                                         @@: 
2019-12-31 10:20:41.656 7719-7719/com.example.sourcecodetest D/Retrofit: zwm, onComplete, thread: main