ROOM+RxJava实现用户存在校验与数据插入问题求助

嘿，我完全懂你现在的困扰——刚接触RxJava和Room的时候，很容易被异步特性坑到，你写的existsUser方法里，return语句确实会比RxJava的回调先执行，所以永远拿不到正确的用户存在状态。咱们来一步步把这个问题搞定！

为什么原来的代码行不通？ #

RxJava的subscribe是异步执行的：当你调用db.userDao().existsUser(...).subscribe(...)时，查询操作会被放到Schedulers.io()线程去跑，而主线程会直接往下走执行return existe[0]，这时候回调里的existe[0] = query > 0还没执行，所以永远返回初始值false。

方案1：用RxJava的流串联整个逻辑（推荐） #

既然用了RxJava，咱们就顺着它的异步思路来，把校验和插入的逻辑都放到流里处理，这样能保证顺序和正确性。

第一步：修改UserRepository的方法

把existsUser改成返回Single<Boolean>，让调用方可以订阅这个结果；同时把insert改成返回Completable，方便上层统一处理流：

public class UserRepository {
    private AppDatabase db;

    public UserRepository(Context context){
        this.db = AppDatabase.getDatabase(context);
    }

    // 改成返回Single<Boolean>，直接映射查询结果为布尔值
    public Single<Boolean> existsUser(User item){
        return db.userDao().existsUser(item.getName(), item.getEmail())
                .map(count -> count > 0)
                .subscribeOn(Schedulers.io());
    }

    // 改成返回Completable，不要在内部订阅，交给上层处理
    public Completable insertUser(User user){
        return db.userDao().insert(user)
                .subscribeOn(Schedulers.io()); // 数据库操作放在io线程更合适
    }
}

第二步：修改Activity的校验逻辑

用RxJava的操作符遍历用户列表，依次校验并插入不存在的用户：

UserRepository repo = new UserRepository(LoadingActivity.this);

Observable.fromIterable(list)
        // 依次处理每个用户，保证顺序
        .concatMap(user -> repo.existsUser(user)
                // 只保留不存在的用户
                .filter(exists -> !exists)
                // 把布尔值转回用户对象，方便后续插入
                .map(exists -> user)
        )
        // 依次插入每个符合条件的用户
        .concatMapCompletable(user -> repo.insertUser(user))
        // 整个逻辑放在io线程执行
        .subscribeOn(Schedulers.io())
        // 结果回调回到主线程
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new CompletableObserver() {
            @Override
            public void onSubscribe(Disposable d) {
                // 这里可以显示加载提示，比如ProgressBar
            }

            @Override
            public void onComplete() {
                // 所有用户处理完成，更新UI或者跳转页面
                Toast.makeText(LoadingActivity.this, "用户同步完成", Toast.LENGTH_SHORT).show();
            }

            @Override
            public void onError(Throwable e) {
                // 处理错误，比如显示错误提示
                Toast.makeText(LoadingActivity.this, "同步失败：" + e.getMessage(), Toast.LENGTH_SHORT).show();
            }
        });

方案2：用blockingGet同步获取结果（不推荐主线程使用） #

如果你暂时不想大改上层逻辑，也可以用blockingGet()来同步获取查询结果，但绝对不能在主线程调用，否则会导致ANR：

public class UserRepository {
    // 注意：这个方法必须在子线程调用！
    public boolean existsUserBlocking(User item){
        try {
            Long count = db.userDao().existsUser(item.getName(), item.getEmail())
                    .subscribeOn(Schedulers.io())
                    .blockingGet();
            return count > 0;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }
}

然后在Activity里用RxJava切换到io线程执行：

UserRepository repo = new UserRepository(LoadingActivity.this);

Observable.fromIterable(list)
        .subscribeOn(Schedulers.io())
        .forEach(user -> {
            if (!repo.existsUserBlocking(user)) {
                repo.insert(user);
            }
        })
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(() -> {
            // 处理完成
        }, throwable -> {
            // 处理错误
        });

为什么不推荐方案2？ #

blockingGet()会阻塞当前线程，虽然能拿到同步结果，但违背了RxJava异步编程的初衷，而且如果处理大量用户，可能会导致线程长时间被占用，影响性能。

额外小建议 #

你提到不想用@Insert(onConflict = OnConflictStrategy.IGNORE)，因为数据没有主键——其实可以考虑把name和email设为复合主键，这样Room就能自动处理冲突了，这也是一种更简洁的方案：

@Entity(primaryKeys = {"name", "email"})
public class User {
    private String name;
    private String email;
    // 其他字段和Getter/Setter
}

这样之后直接用@Insert(onConflict = OnConflictStrategy.IGNORE)，就能自动跳过已存在的用户，不用自己写校验逻辑了，不过这取决于你的业务是否允许把这两个字段作为主键。

内容的提问来源于stack exchange，提问作者José Ángel Buesa