Realm 中的线程 - 干净的代码架构

Question

我在 Android 上使用 Clean Code Architecture 和 Dagger with Realm，但我找不到让它一起工作的方法。 底线是，我总是得到：

java.lang.IllegalStateException: Realm access from incorrect thread. Realm objects can only be accessed on the thread they were created.

FetchItemsRepo:

 @Override
    public Observable<List<Item>> fetchAllItems() {
        final List<Item> items = new ArrayList<>();
        final Realm realm = Realm.getInstance(new RealmConfiguration.Builder(context).build());

        final RealmQuery<Checkboxes> queryCheck = realm.where(Checkboxes.class);
        final RealmQuery<ImageItem> queryImage = realm.where(ImageItem.class);

        return queryCheck.findAll()
                .asObservable()
                .flatMap(new Func1<RealmResults<Checkboxes>, Observable<RealmResults<ImageItem>>>() {
                    @Override
                    public Observable<RealmResults<ImageItem>> call(RealmResults<Checkboxes> checkboxes) {
                        for (Checkboxes checks : checkboxes)
                            items.add(checks);

                        return queryImage.findAll().asObservable();
                    }
                })
                .flatMap(new Func1<RealmResults<ImageItem>, Observable<List<Item>>>() {
                    @Override
                    public Observable<List<Item>> call(RealmResults<ImageItem> imageItems) {
                        for (ImageItem images : imageItems)
                            items.add(images);

                        return Observable.create(new Observable.OnSubscribe<List<Item>>() {
                            @Override
                            public void call(Subscriber<? super List<Item>> subscriber) {
                                subscriber.onNext(items);
                            }
                        });
                    }
                });
    }

GetItemsUseCase:

public class GetItemsUseCase extends UseCase {

    private final ItemsRepository itemsRepository;

    @Inject
    public GetItemsUseCase(ItemsRepository itemsRepository, ThreadExecutor threadExecutor, PostExecutionThread postExecutionThread) {
        super(threadExecutor, postExecutionThread);
        this.itemsRepository = itemsRepository;
    }

    @Override
    protected Observable buildUseCaseObservable() {
        return this.itemsRepository.fetchAllItems();
    }
}

还有UseCase：

public void execute(Subscriber UseCaseSubscriber) {
        this.subscription = this.buildUseCaseObservable()
                .subscribeOn(Schedulers.from(threadExecutor))
                .observeOn(postExecutionThread.getScheduler())
                .subscribe(UseCaseSubscriber);
    }

我不确定是否有任何方法可以让 Realm 使用这种模式。 整个项目在我的GitHub上开源：https://github.com/leonardo2204/materialnotes/tree/bug_thread_realm

编辑：

Realm access from incorrect thread. Realm objects can only be accessed on the thread they were created.
 java.lang.IllegalStateException: Realm access from incorrect thread. Realm objects can only be accessed on the thread they were created.
 at io.realm.BaseRealm.checkIfValid(BaseRealm.java:456)
 at io.realm.RealmResults.addChangeListener(RealmResults.java:926)
 at io.realm.rx.RealmObservableFactory$5.call(RealmObservableFactory.java:147)
 at io.realm.rx.RealmObservableFactory$5.call(RealmObservableFactory.java:131)
 at rx.Observable.unsafeSubscribe(Observable.java:9860)
 at rx.internal.operators.OnSubscribeMap.call(OnSubscribeMap.java:48)
 at rx.internal.operators.OnSubscribeMap.call(OnSubscribeMap.java:33)
 at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:48)
 at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:30)
 at rx.Observable.unsafeSubscribe(Observable.java:9860)
 at rx.internal.operators.OnSubscribeMap.call(OnSubscribeMap.java:48)
 at rx.internal.operators.OnSubscribeMap.call(OnSubscribeMap.java:33)
 at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:48)
 at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:30)
 at rx.Observable.unsafeSubscribe(Observable.java:9860)
 at rx.internal.operators.OperatorSubscribeOn$1.call(OperatorSubscribeOn.java:94)
 at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55)
 at rx.internal.schedulers.ExecutorScheduler$ExecutorSchedulerWorker.run(ExecutorScheduler.java:104)
 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1112)
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:587)
 at java.lang.Thread.run(Thread.java:841)

Answer 1

您需要使用 realm.copyFromRealm() 使您的 Realm 对象在您的 RealmResults 中分离。

编辑：它说您的问题是您的 Realm 实例与您订阅的线程不在同一个线程上，如果您订阅 Schedulers.from(threadExecutor) 但您在 UI 线程上初始化此查询，这是公平的。

asObservable() 无论如何都不能在后台线程上工作，只能在 UI 线程上工作 - 因为它需要添加一个更改侦听器，而更改侦听器需要主循环器来接收更新。

老实说，按照这个速度，您可以这样做（并失去自动更新）：

    final List<Item> items = new ArrayList<>();
    Realm realm = null;
    try {
         realm = Realm.getInstance(new RealmConfiguration.Builder(context).build());

        final RealmQuery<Checkboxes> queryCheck = realm.where(Checkboxes.class);
        final RealmQuery<ImageItem> queryImage = realm.where(ImageItem.class);

        RealmResults<Checkboxes> checkBoxes = queryCheck.findAll();
        RealmResults<ImageItem> imageItems = queryImage.findAll();

        for(Checkboxes checkbox : checkBoxes) {
            items.add(realm.copyFromRealm(checkbox));
        }
        for(ImageItem checkbox : checkBoxes) {
            items.add(realm.copyFromRealm(checkbox));
        }
    } finally {
        if(realm != null) {
            realm.close();
        }
    }
    return Observable.just(items);

或者只是重新考虑您执行此操作的方式，因为最佳解决方案是使用 UI 线程中的findAllAsync()，如果您可以在一个结果列表中获取两个列表。

EDIT2：值得注意的是，asObservable() 的预期用途是替换 addChangeListener()，但您只能在 Looper 线程（通常是 UI 线程）上“观察”Realm 的变化。

所以asObservable() 的预期用法如下

private Subscription readFromEditText() {
    return RxTextView.textChanges(editText).switchMap(charSequence -> {
        String selectedName = charSequence.toString();
        RealmQuery<Dog> query = realm.where(Dog.class);
        if(selectedName != null && !"".equals(selectedName)) {
            query = query.contains(DogFields.NAME, selectedName, Case.INSENSITIVE);
        }
        return query.findAllSortedAsync(DogFields.NAME)
                    .asObservable();
    }).filter(RealmResults::isLoaded) //filter async realm query
      .subscribe(dogs -> adapter.updateData(dogs));
}