使用copyFromRealm发送副本时,Realm访问错误的线程异常

Question

当流式处理领域对象的副本而不是领域引用并在Schedulers.IO线程上观察它时,会发生着名的异常消息"Realm从不正确的线程访问."Realm对象只能在它们创建的线程中访问.

该副本不应该是无线程的吗？我可以从一个线程生成它并在不同的线程上处理它吗？

这就是我创建可观察的方式.

 public Observable<Brand> getAllBrands() {
    return realm.where(Brand.class)
            .findAll()
            .asObservable()
            .flatMap(Observable::from)
            .map(brand -> realm.copyFromRealm(brand));
}

以下是我观察getAllBrands()的方法.

 Observable<Brand> brandObservable = dataManager.getAllBrands();

    brandObservable.observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .subscribe(new Observer<Brand>() {
                @Override
                public void onCompleted() {
                    Log.d("reactive", "completed");
                }

                @Override
                public void onError(Throwable e) {
                    Log.d("reactive", e.getMessage());
                }

                @Override
                public void onNext(Brand brand) {
                    dataSource.add(brand.getName());
                    myAdapter.notifyDataSetChanged();
                }
            });

Answer 1

您正在schedulers.io使用UI线程中的Realm实例进行订阅:

realm.where(Brand.class)
        .findAll()
        .asObservable()
        .flatMap(Observable::from)
        .map(brand -> realm.copyFromRealm(brand)) // realm instance on the wrong thread
        .subscribeOn(schedulers.io());            

你在追求什么是一种简单的方法来跨线程移动查询,这仍然是在进行中的工作:https://github.com/realm/realm-java/pull/1978.在此之前,您可以通过以下方式自行解决:

public Observable<Brand> getAllBrands(final Realm realm) {
    return Observable.create(new Observable.OnSubscribe<List<Brand>>() {
        @Override
        public void call(final Subscriber<? super List<Brand>> subscriber) {
            Realm obsRealm = Realm.getInstance(realm.getConfiguration());
            final RealmResults<Brand> results = obsRealm.where(Brand.class).findAll();
            final RealmChangeListener listener = new RealmChangeListener() {
                @Override
                public void onChange() {
                    subscriber.onNext(realm.copyFromRealm(results));
                }
            };
            results.addChangeListener(listener);
            subscriber.add(Subscriptions.create(new Action0() {
                @Override
                public void call() {
                    realm.removeChangeListener(listener);
                    realm.close();
                }
            }));

        }
    })
    .flatMap(Observable::from);
}

请注意,Realm Changelisteners仅适用于Looper线程,这意味着您可能需要将工作线程更改为H

HandlerThread bgThread = new HandlerThread("workerThread");
Handler handler = new Handler(bgThread.getLooper());

getAllBrands(realm).subscribeOn(HandlerScheduler.from(handler));

Answer 2

问题

我可以从一个线程生成它并在另一个线程上处理它吗？

答案是否定的。您可以在这里阅读领域的描述

虽然 Realm 文件可以被多个线程并发访问，但是您不能在线程之间移交 Realm、Realm 对象、查询和结果。线程示例展示了如何在多线程环境中使用 Realm。