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

Question

我正在 Android 上使用干净代码架构，并使用 Dagger 和 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() 在 RealmResults 中分离 Realm 对象。

编辑:它说你的问题是你的 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上“观察”Realm的变化线程(通常是 UI 线程)。

因此 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));
}