在 Rx Observable 事务中使用领域调用时，如何防止领域线程问题？

Question

这是调用 Rx 事务的 ViewModel

RealmHelperRepo 是 HelperRepo 接口的实现

@PerActivity
public class RoleSelectionViewModel extends BaseViewModel<RoleSelectionMvvm.View> implements RoleSelectionMvvm.ViewModel {

     private Disposable roleGroupSubscription;

     @Inject
        public RoleSelectionViewModel(@AppContext Context context, HelperRepo helperRepo, ApiOAuth2 ApiOAuth2) {

        this.mContext = context;
        this.mUserRepo = userRepo;
        this.mHelperRepo = helperRepo;
        ApiOAuth2.initialize();
        this.mApiOAuth2 = ApiOAuth2;

        this.mCurrentUser = mUserRepo.getByField("isLoggedIn", true, true);
        if (mCurrentUser != null) {
            this.mCurrentUserId = mCurrentUser.getId();
            this.mHelper = mHelperRepo.getByField("user.id", mCurrentUserId, true);
    }

  Observable<Response<ResponseHelper>> postHelperObservable = mApiOAuth2.postHelperRX(new Helper());
  Observable<Response<ResponseHelper>> getHelperObservable = mApiOAuth2.getHelperRX(mCurrentUserId);

 roleGroupSubscription = postRoleGroupsObservable
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .flatMap((response) -> {
                        if (response.isSuccessful()) {
                            ResponseHelper responseHelper = response.body();
                            mHelper = responseHelper.getHelper();
                            return Observable.just(mHelper);
                        } else if (response.code() == 409) {
                       // handle POST conflict (i.e. helper already exists)
                            return getHelperObservable;
                        }

                   })
                   .subscribe((data) -> {
                        if (data instanceof Response<?>) {
                            // data came from getHelperObservable
                            Response response = (Response) data;
                            if (!response.isSuccessful()) {
                               ResponseHelper responseHelper = (ResponseHelper) response.body();
                            mHelper = responseHelper.getHelper();
                       else {

                            // data came from Observable.just(helper)
                            mApiOAuth2.getHelperRX(mCurrentUserId).subscribe(
                                    responseHelperResponse -> {

                                        if (responseHelperResponse.isSuccessful()) {

                                            String helperID = responseHelperResponse.body().getHelper().getId();
                                            Log.d("RealmCount", "save: " + Realm.getLocalInstanceCount(realmProvider.get().getConfiguration()));
                                            mHelper.setId(helperID);
                                            mHelper.setUser(mCurrentUser);
--------> // when mHelperRepo.save(mHelper) is called, it goes to RealmHelperRepo to save and 
--------> // thus triggering mRealm.executeTransaction causing Realm threading
                                            mHelperRepo.save(mHelper);
                                        }
                                        saveAndBegin();
                                    },
                                    Throwable::printStackTrace);
                                    });

这是进行领域调用的 RealmRepo 类。

@PerApplication
public class RealmHelperRepo implements HelperRepo {

   private final Provider<Realm> mRealmProvider;
   private Realm mRealm;

   @Inject
    public RealmHelperRepo(Provider<Realm> realmProvider) {
        this.mRealmProvider = realmProvider;
        this.mRealm = mRealmProvider.get();
}


  @Override
    public void save(Helper helper) {
        if (mRealm != null) {
---------> // code runs into threading issue here when a realmThread executeTransaction is called
        mRealm.executeTransaction(r -> r.copyToRealmOrUpdate(helper)); 
        }
    }

这里有什么我遗漏的吗？我应该使用其他 Rx 函数而不是平面图？还有其他方法可以保存我的可观察数据而不会遇到线程问题吗？帮忙！

Answer 1

这里有什么我遗漏的吗？

Realm 实例表示引用计数、线程本地 实例。这不是一个全局的东西，它是一个“本地实例”，由getInstance() 打开，然后由close() 关闭。

因此，您不能只将 Realm 实例初始化为单例，因为后台线程无法访问它。

---

例如，您可以提供一个能够打开线程本地 Realm 实例的单例 Realm 管理器类。

/**
 * The RealmManager allows creating a singleton Realm manager which can open thread-local instances.
 *
 * It also allows obtaining the open thread-local instance without incrementing the reference count.
 */
@PerApplication
public class RealmManager {
    private final ThreadLocal<Realm> localRealms = new ThreadLocal<>();

    @Inject
    RealmManager() {
    }

    /**
     * Opens a reference-counted local Realm instance.
     *
     * @return the open Realm instance
     */
    public Realm openLocalInstance() {
        checkDefaultConfiguration();
        Realm realm = Realm.getDefaultInstance(); // <-- maybe configuration should be constructor parameter
        if(localRealms.get() == null) {
            localRealms.set(realm);
        }
        return realm;
    }

    /**
     * Returns the local Realm instance without adding to the reference count.
     *
     * @return the local Realm instance
     * @throws IllegalStateException when no Realm is open
     */
    public Realm getLocalInstance() {
        Realm realm = localRealms.get();
        if(realm == null) {
            throw new IllegalStateException(
                    "No open Realms were found on this thread.");
        }
        return realm;
    }

    /**
     * Closes local Realm instance, decrementing the reference count.
     *
     * @throws IllegalStateException if there is no open Realm.
     */
    public void closeLocalInstance() {
        checkDefaultConfiguration();
        Realm realm = localRealms.get();
        if(realm == null) {
            throw new IllegalStateException(
                    "Cannot close a Realm that is not open.");
        }
        realm.close();
        // noinspection ConstantConditions
        if(Realm.getLocalInstanceCount(Realm.getDefaultConfiguration()) <= 0) {
            localRealms.set(null);
        }
    }

    private void checkDefaultConfiguration() {
        if(Realm.getDefaultConfiguration() == null) {
            throw new IllegalStateException("No default configuration is set.");
        }
    }
}

你可以这样使用

@PerApplication
public class RealmHelperRepo implements HelperRepo {
   private final RealmManager realmManager;

    @Inject
    public RealmHelperRepo(RealmManager realmManager) {
        this.realmManager = realmManager;
    }


    @Override
    public void save(Helper helper) {
        try(Realm realm = realmManager.openLocalInstance()) {
            realm.executeTransaction(r -> r.copyToRealmOrUpdate(helper)); 
        }
    }

从技术上讲，它只是隐藏了 Realm.getDefaultInstance() 调用，并允许您在不增加内部 RealmCache 引用计数的情况下获取线程本地实例，因此没有太多真正的魔力。

只需为线程打开一个 Realm 实例，不要忘记在不再需要时关闭它。