RxJava2中postRegisterAccount方法内subscriber为null的原因排查

我正在学习RxJava，排查问题时遇到了这个错误：API请求成功，但在Single<ResponseRegistration> postRegisterAccount()方法中出现错误："onSuccess called with null. Null values are generally not allowed in 2.x operators"。同时发现该方法中的subscriber为null。

相关代码如下：

View层代码 #

addRxSubscription(mService.postRegistration()
 .subscribeOn(Schedulers.io())
 .observeOn(AndroidSchedulers.mainThread())
 .subscribe(success -> {
 if (success != null && success) {
 //successfully registered
 updateNavigationDrawerMenu();
 } else {
 Log.e(TAG, "Failed to register !");
 }
 }, throwable -> Log.e(TAG, "Failed to register !")
));
break;

Service层代码 #

public Single<Boolean> postRegistration() {
 final String email = "test@email.com";
 final String pass = "testpass";
 PostRegistrationData registrationData = new PostRegistrationData(email, pass);
 Single<ResponseRegistration> singlePostRegistration = mApi.postRegisterAccount(registrationData);
 return Single.create(subscriber -> singlePostRegistration.subscribeOn(Schedulers.io())
 .observeOn(Schedulers.io())
 .subscribe(registrationResponse -> {
 if (registrationResponse.userId != null) {
 // Success, store the response and let UI know API was successful
 storeRegistrationId(registrationResponse.userId);
 if (!subscriber.isDisposed()) {
 subscriber.onSuccess(true);
 }
 } else {
 // Failure, let UI know API was un-successful
 if (!subscriber.isDisposed()) {
 subscriber.onSuccess(false);
 }
 }
 }, throwable -> {
 // Failure
 if (!subscriber.isDisposed()) {
 // // Due to the failure in postRegisterAccount(), this code executes
 // subscriber.onError(throwable);
 }
 }));
}

API层代码 #

public Single<ResponseRegistration> postRegisterAccount(PostRegistrationData registrationData) {
 Single<Response<ResponseGeneric<ResponseRegistration>>> postRegistration = apiClient.postRegisterAccount(registrationData);
 return Single.create(subscriber -> postRegistration.subscribeOn(Schedulers.io())
 .observeOn(Schedulers.io())
 .subscribe(response -> {
 if(response.isSuccessful() && response.body() != null) {
 // // this part is successful - the response data is received
 // HOWEVER java.lang.NullPointerException: onSuccess called with null. Null values are generally not allowed in 2.x operators and sources.
 // subscriber = null
 // if (!subscriber.isDisposed()) subscriber.onSuccess(response.body().data);
 }
 }, throwable -> {
 throwable.printStackTrace();
 if(!subscriber.isDisposed()) subscriber.onError(throwable);
 }));
}

Retrofit接口 #

@Headers({"Content-Type: application/json"})
@POST(ServicesConstants.API_PREFIX + "registration/")
Single<Response<ResponseGeneric<ResponseRegistration>>> postRegisterAccount(@Body PostRegistrationData postRegistrationData);

请问：为何postRegisterAccount()方法中的subscriber会为null？

核心原因：错误使用Single.create()导致的生命周期不匹配 #

你遇到的subscriber为null的问题，本质是不必要地嵌套Single.create()并手动管理订阅逻辑，违反了RxJava的设计原则：

• Single.create()的回调时机问题：
  当你在API层用Single.create()包裹postRegistration的订阅时，postRegistration是Retrofit返回的Single，它会在被订阅时立即发起网络请求。如果postRegistration的回调（网络请求完成）比外部订阅你创建的这个Single更早执行，此时Single.create()传入的subscriber参数还未被初始化（外部还没触发订阅），自然就会出现subscriber为null的情况。

• 重复的线程调度毫无意义：
  RxJava中subscribeOn()只会生效一次（决定事件发射的线程），你在Service层、API层、View层多次调用subscribeOn(Schedulers.io())完全冗余，反而增加了代码复杂度和潜在的线程问题。

• 手动订阅嵌套Single的风险：
  在Single.create()内部手动订阅另一个Single，会让你失去RxJava自动管理的订阅生命周期，很容易出现subscriber未绑定、内存泄漏或者事件丢失的问题。

修复方案：用RxJava标准操作符替代Single.create() #

我们可以把所有手动创建Single的逻辑替换为map()、flatMap()等标准操作符，让RxJava自动处理订阅生命周期：

1. 重构API层代码

public Single<ResponseRegistration> postRegisterAccount(PostRegistrationData registrationData) {
    return apiClient.postRegisterAccount(registrationData)
            .map(response -> {
                if (response.isSuccessful() && response.body() != null) {
                    ResponseRegistration data = response.body().data;
                    if (data == null) {
                        // 处理data为null的情况，抛出错误避免onSuccess传null
                        throw new NullPointerException("Response data is null");
                    }
                    return data;
                } else {
                    // 处理请求失败的情况，抛出自定义异常
                    throw new IOException("Request failed with code: " + response.code());
                }
            });
}

2. 重构Service层代码

public Single<Boolean> postRegistration() {
    final String email = "test@email.com";
    final String pass = "testpass";
    PostRegistrationData registrationData = new PostRegistrationData(email, pass);
    
    return mApi.postRegisterAccount(registrationData)
            .map(registrationResponse -> {
                if (registrationResponse.userId != null) {
                    storeRegistrationId(registrationResponse.userId);
                    return true;
                } else {
                    return false;
                }
            })
            // 处理API层抛出的异常
            .onErrorReturn(throwable -> {
                Log.e(TAG, "Registration API error", throwable);
                return false;
            });
}

3. View层代码保持简洁（仅保留必要的线程调度）

addRxSubscription(mService.postRegistration()
        .subscribeOn(Schedulers.io()) // 只在这里指定一次事件发射线程
        .observeOn(AndroidSchedulers.mainThread()) // 指定UI线程接收结果
        .subscribe(success -> {
            if (success) {
                updateNavigationDrawerMenu();
            } else {
                Log.e(TAG, "Failed to register !");
            }
        }, throwable -> Log.e(TAG, "Failed to register !", throwable)));
break;

额外注意事项 #

• RxJava 2.x禁止onSuccess传递null：如果你的API返回的data可能为null，一定要提前处理（比如抛出异常或者返回默认值），避免触发NullPointerException。
• 使用CompositeDisposable管理订阅：确保在页面销毁时清理所有RxJava订阅，避免内存泄漏（你的addRxSubscription应该是做这个的，继续保持）。
• 避免手动创建Single/Observable：除非你需要实现自定义的事件发射逻辑，否则尽量用标准操作符组合，这是RxJava的最佳实践。

内容的提问来源于stack exchange，提问作者Someone Somewhere