使用具有活动对象和内部领域实例处理的领域的存储库模式

Question

我正在尝试实现一个使用领域的存储库模式，并且在处理领域实例的创建和关闭时仍然保留活动对象功能。这是我目前的方法，遗憾的是它不起作用。我在我的自定义后台线程上运行所有这些代码。

public Observable<List> getAll(Class clazz) {
    Realm realm = Realm.getDefaultInstance();
    SerializedSubject relay = new SerializedSubject<>(PublishSubject.create());
    try {
         realm.where(clazz).findAllAsync().addChangeListener(new RealmChangeListener<RealmResults>() {
               @Override
               public void onChange(RealmResults element) {
                      relay.onNext(realm.copyFromRealm(element));
               }
         });
         return relay.asObservable();
    } finally {
         relay.onCompleted();
         try {
            if (!realm.isClosed())
                realm.close();
        } catch (IllegalStateException e) {
            e.printStackTrace();
        }
    }
}

后台线程：

public class JobExecutor implements ThreadExecutor {

private static final int INITIAL_POOL_SIZE = Runtime.getRuntime().availableProcessors();
// Sets the amount of time an idle thread waits before terminating
private static final int KEEP_ALIVE_TIME = 10;
// Sets the Time Unit to seconds
private static final TimeUnit KEEP_ALIVE_TIME_UNIT = TimeUnit.SECONDS;
@NonNull
private final ThreadPoolExecutor threadPoolExecutor;

public JobExecutor() {
    threadPoolExecutor = new ThreadPoolExecutor(INITIAL_POOL_SIZE, INITIAL_POOL_SIZE,
            KEEP_ALIVE_TIME, KEEP_ALIVE_TIME_UNIT, new LinkedBlockingQueue<>(), new JobThreadFactory());
}

@Override
public void execute(@NonNull Runnable runnable) {
    this.threadPoolExecutor.execute(runnable);
}

private static class JobThreadFactory implements ThreadFactory {
    private static final String THREAD_NAME = "android_";
    private int counter = 0;

    @NonNull
    @Override
    public Thread newThread(@NonNull Runnable runnable) {
        return new Thread(runnable, THREAD_NAME + counter++);
    }
}

我是这样使用的：

private <T> Observable.Transformer<T, T> applySchedulers() {
    return observable -> observable.subscribeOn(Schedulers.from(mThreadExecutor)) // my background thread
            .observeOn(mPostExecutionThread.getScheduler()); // main thread
}

抛出此异常：

java.lang.IllegalStateException: Your Realm is opened from a thread without a Looper. Async queries need a Handler to send results of your query

当我使用 .findAll() 而不是 .findAllAsync() 时会引发此异常：

java.lang.IllegalStateException: You can't register a listener from a non-Looper thread or IntentService thread. 

提前致谢，期待您的回复。

Answer 1

您需要一个具有 Looper/Handler 的线程，如 IllegalStateException 所述。现在，看起来您只是在使用普通的 Schedulers 类，而不是 AndroidScheduler。查看此针对 Android 特定调度程序的存储库。

https://github.com/ReactiveX/RxAndroid/tree/1.x/rxandroid/src/main/java/rx/android/schedulers