【3天精通】RxJava+Retrofit实战：从0到1打造响应式网络请求框架

2026-01-18 10:28:56作者：苗圣禹Peter

引言：还在为Android网络请求头疼？

你是否遇到过这些问题：

网络请求与UI线程纠缠不清导致ANR（Application Not Responding，应用无响应）
重复封装相同格式的HTTP请求代码
无法优雅地取消正在进行的网络请求
异步回调嵌套形成"回调地狱"（Callback Hell）

本文将通过RxjavaRetrofitDemo开源项目，手把手教你实现一个生产级别的响应式网络请求框架。读完本文你将掌握：
✅ RxJava与Retrofit的深度整合方案
✅ 网络请求的统一封装与错误处理
✅ 带进度对话框的请求订阅器实现
✅ 网络请求的生命周期管理与取消机制

技术栈选型与环境准备

核心依赖组件

组件	版本	作用
Retrofit	2.0+	类型安全的HTTP客户端
RxJava	1.x	异步事件流处理库
OkHttp	3.x	HTTP请求底层实现
Gson	2.8+	JSON数据解析

环境搭建步骤

克隆项目

git clone https://gitcode.com/gh_mirrors/rx/RxjavaRetrofitDemo

项目结构解析

RxjavaRetrofitDemo/
├── app/
│   └── src/main/java/com/queen/rxjavaretrofitdemo/
│       ├── http/           # 网络请求核心模块
│       │   ├── HttpMethods.java      # 请求管理类
│       │   ├── MovieService.java     # API接口定义
│       │   └── ResponseConvertFactory.java  # 响应转换器
│       └── subscribers/    # 订阅器实现
│           └── ProgressSubscriber.java  # 带进度对话框的订阅器

核心功能实现详解

1. 单例模式的网络请求管理类

HttpMethods.java采用双重校验锁单例模式，确保全局只有一个Retrofit实例，避免重复创建造成的资源浪费：

private static class SingletonHolder{
    private static final HttpMethods INSTANCE = new HttpMethods();
}

public static HttpMethods getInstance(){
    return SingletonHolder.INSTANCE;
}

Retrofit配置关键代码：

retrofit = new Retrofit.Builder()
    .client(builder.build())
    .addConverterFactory(ResponseConvertFactory.create()) // 自定义响应转换器
    .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) // RxJava适配
    .baseUrl(BASE_URL)
    .build();

2. API接口定义与请求封装

2.1 定义API服务接口

MovieService.java使用Retrofit注解定义HTTP请求：

public interface MovieService {
    @GET("top250")
    Observable<HttpResult<List<Subject>>> getTopMovie(
        @Query("start") int start, 
        @Query("count") int count
    );
}

2.2 请求参数与响应统一封装

请求参数标准化：
所有分页请求统一使用start（起始位置）和count（请求数量）参数。

响应数据格式：

public class HttpResult<T> {
    private int count;          // 数据总数
    private int start;          // 起始索引
    private List<T> subjects;   // 数据列表
    
    // Getter/Setter省略
}

3. 响应数据预处理与错误处理

3.1 统一响应转换器

ResponseConvertFactory实现对HTTP响应的统一预处理：

public class ResponseConvertFactory extends Converter.Factory {
    // 自定义Gson响应转换器
    private final GsonConverterFactory gsonConverterFactory = GsonConverterFactory.create();
    
    @Override
    public Converter<ResponseBody, ?> responseBodyConverter(Type type, Annotation[] annotations, Retrofit retrofit) {
        // 获取默认Gson转换器
        Converter<ResponseBody, ?> converter = gsonConverterFactory.responseBodyConverter(type, annotations, retrofit);
        return new GsonResponseBodyConverter((Converter<ResponseBody, HttpResult>) converter);
    }
}

3.2 结果处理函数

HttpMethods中定义HttpResultFunc处理响应结果：

private class HttpResultFunc<T> implements Func1<HttpResult<T>, T> {
    @Override
    public T call(HttpResult<T> httpResult) {
        if (httpResult.getCount() == 0) {
            throw new ApiException(100); // 自定义异常码
        }
        return httpResult.getSubjects();
    }
}

4. 带进度对话框的订阅器实现

ProgressSubscriber.java实现了网络请求的可视化管理：

public class ProgressSubscriber<T> extends Subscriber<T> implements ProgressCancelListener {
    private ProgressDialogHandler mProgressDialogHandler;
    
    @Override
    public void onStart() {
        showProgressDialog(); // 订阅开始时显示对话框
    }
    
    @Override
    public void onCompleted() {
        dismissProgressDialog(); // 请求完成时关闭对话框
        Toast.makeText(context, "Get Top Movie Completed", Toast.LENGTH_SHORT).show();
    }
    
    @Override
    public void onError(Throwable e) {
        dismissProgressDialog();
        if (e instanceof SocketTimeoutException) {
            Toast.makeText(context, "网络中断，请检查您的网络状态", Toast.LENGTH_SHORT).show();
        } else {
            Toast.makeText(context, "error:" + e.getMessage(), Toast.LENGTH_SHORT).show();
        }
    }
    
    @Override
    public void onCancelProgress() {
        if (!this.isUnsubscribed()) {
            this.unsubscribe(); // 取消订阅即取消请求
        }
    }
}

5. 完整请求流程实现

5.1 请求调度流程

sequenceDiagram
    participant Activity
    participant HttpMethods
    participant Retrofit
    participant MovieService
    participant ProgressSubscriber
    
    Activity->>HttpMethods: getInstance()
    HttpMethods->>HttpMethods: 创建单例
    Activity->>HttpMethods: getTopMovie(subscriber, 0, 10)
    HttpMethods->>MovieService: getTopMovie(0,10)
    MovieService->>Retrofit: 发起HTTP请求
    Retrofit->>HttpMethods: 返回Observable<HttpResult>
    HttpMethods->>HttpResultFunc: 转换结果
    HttpResultFunc->>ProgressSubscriber: 返回数据列表
    ProgressSubscriber->>Activity: 回调onNext()

5.2 调用示例

在Activity中使用：

// 创建订阅者
Subscriber<List<Subject>> subscriber = new ProgressSubscriber<>(
    new SubscriberOnNextListener<List<Subject>>() {
        @Override
        public void onNext(List<Subject> subjects) {
            // 处理电影列表数据
            updateUI(subjects);
        }
    }, 
    this // Activity上下文
);

// 发起请求
HttpMethods.getInstance().getTopMovie(subscriber, 0, 20);

6. 网络请求的取消机制

实现请求取消的两种方式：

通过订阅器取消

// 在Activity的onDestroy中
if (subscriber != null && !subscriber.isUnsubscribed()) {
    subscriber.unsubscribe();
}

通过ProgressDialog取消

@Override
public void onCancelProgress() {
    if (!this.isUnsubscribed()) {
        this.unsubscribe(); // 取消订阅即取消请求
    }
}

高级应用与最佳实践

请求生命周期管理

生命周期方法	操作
onCreate()	创建订阅器
onStart()	发起请求
onPause()	取消订阅
onDestroy()	清理资源

错误处理策略

@Override
public void onError(Throwable e) {
    if (e instanceof SocketTimeoutException) {
        showError("网络超时");
    } else if (e instanceof ConnectException) {
        showError("连接失败");
    } else if (e instanceof ApiException) {
        showError("数据异常: " + ((ApiException)e).getCode());
    } else {
        showError("未知错误");
    }
}

性能优化建议

添加请求缓存

OkHttpClient.Builder builder = new OkHttpClient.Builder()
    .addNetworkInterceptor(new CacheInterceptor())
    .cache(new Cache(cacheDir, cacheSize));

请求合并与批处理
使用Observable.merge()合并多个请求：

Observable.merge(request1, request2)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(combinedSubscriber);