java - 使用 RxJava 在同一后台线程中转换 RealmResult

Question

我正在使用 RxJava + Realm 并想做一些看起来非常简单的事情:

我想加载所有符合特定条件的 RealmObject，转换每个 RealmResult 对象，然后在这些转换后的对象准备好后通知主线程。我需要在后台线程中进行转换的原因，以及如果 RealmResults 中有大量对象，则这是一个长时间运行的操作

问题是，我可以在后台加载对象，但仍处于后台线程中时无法对每个对象执行任何操作。我也在尝试以被动的方式做到这一点。

到目前为止的代码:

groupContactSubscription = realm.where(GroupContact.class)
                .equalTo(MODEL_ID, mGroupId)
                .findAllAsync()
                .asObservable()
                .filter(new Func1<RealmResults<GroupContact>, Boolean>() {
                    @Override
                    public Boolean call(RealmResults<GroupContact> groupContacts) {
                        return groupContacts.isLoaded();
                    }
                })
                .first()
                .map(new Func1<RealmResults<GroupContact>, List<Contact>>() {
                    @Override
                    public List<Contact> call(RealmResults<GroupContact> groupContacts) {
                        List<Contact> contacts = new ArrayList<>();

                        //Transform each GroupContact
                        for(int i=0; i<groupContacts.size(); ++i){
                          contacts.add(transformGroupContact(groupContacts.get(0))
                        }
                        return contacts;
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<List<Contact>>() {
                    @Override
                    public void call(List<Contact> contacts) {
                        // Deal with transformed objects
                    }
                });

问题是，这不会在后台线程中发生(它仍然阻塞主线程)。

我不太确定应该如何干净地完成此操作，有人可以指出我已经存在的问题的正确方向吗？我相信正在发生的是 Realm 事务是异步发生的，并且过滤器->映射也发生在同一个后台线程上。但显然，事实并非如此。

这是我使用 RxJava 的尝试，所以请友善

Answer 1

       groupContactSubscription = Observable.fromCallable(() -> {
                try(Realm realm = Realm.getDefaultInstance()) {
                    RealmRefresh.refreshRealm(realm); // from http://stackoverflow.com/a/38839808/2413303
                    RealmResults<GroupContact> groupContacts = realm.where(GroupContact.class)
                                                                 .equalTo(MODEL_ID, mGroupId)
                                                                 .findAll();
                    List<Contact> contacts = new ArrayList<>();
                    //Transform each GroupContact
                    for(int i = 0; i < groupContacts.size(); ++i) {
                        contacts.add(transformGroupContact(groupContacts.get(i));
                    }
                    return contacts;                        
                }
            })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<List<Contact>>() {
                @Override
                public void call(List<Contact> contacts) {
                    // Deal with transformed objects
                }
            });

虽然这不会自动更新，所以我对此有点模糊，但我认为这会起作用:

       groupContactSubscription = realm.where(GroupContacts.class).findAll().asObservable()
         .observeOn(Schedulers.io())
         .switchMap(() -> 
             Observable.fromCallable(() -> {
                try(Realm realm = Realm.getDefaultInstance()) {
                    RealmRefresh.refreshRealm(realm); // from http://stackoverflow.com/a/38839808/2413303
                    RealmResults<GroupContact> groupContacts = realm.where(GroupContact.class)
                                                                 .equalTo(MODEL_ID, mGroupId)
                                                                 .findAll();
                    List<Contact> contacts = new ArrayList<>();
                    //Transform each GroupContact
                    for(int i = 0; i < groupContacts.size(); ++i) {
                        contacts.add(transformGroupContact(groupContacts.get(i));
                    }
                    return contacts;                        
                }
            })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Action1<List<Contact>>() {
                @Override
                public void call(List<Contact> contacts) {
                    // Deal with transformed objects
                }
            });

<小时/>

虽然与 a proper Realm schema ，您甚至不需要进行此类转换(这会导致急切地评估整个数据库，将延迟加载抛到窗外)，您只需这样做

Subscription contacts = realm.where(Contact.class)
                                      .equalTo("groupContact.modelId", mGroupId)
                                      .findAllAsync()
                                      .asObservable()
                                      .filter(RealmResults::isLoaded)
                                      .filter(RealmResults::isValid)
                                      .subscribe(...);

或者类似的东西。也许只是按原样返回GroupContact。