如何管理RxJava中需要的loop线程

Question

我想在Observable中封装一个逻辑，该逻辑查询ContentProvider并订阅ContentProvider游标以提供持续更新。

由于可观察对象做IO工作，我需要在Schedulers.io()中订阅它。这样做的问题是，我不能注册ContentObserver，因为它需要一个循环器准备的线程。

建议如何管理并将封装在单个Observable中。

说明代码:

public Observable<Integer> unreadCountObservable() {
    return Observable.create(subscriber -> {
        new UnreadCountObservable(subscriber);
    });
}
private class UnreadCountObservable {
    private Subscriber subscriber;
    public UnreadCountObservable(Subscriber subscriber) {
        this.subscriber = subscriber;
        Cursor cursor = queryUnread(subscriber);
        cursor.registerContentObserver(observer);
        subscriber.add(Subscriptions.create(() -> {
            cursor.unregisterContentObserver(observer);
            cursor.close();
        }));
    }
    @NonNull
    private Cursor queryUnread(Subscriber subscriber) {
        Cursor cursor = contextProvider.getContext().getContentResolver().query(Uri.parse(CONTENT_URI),SMS_PROJECTION,SMS_SELECTION_UNREAD,SMS_PROJECTION,null);
        if(cursor.moveToNext()) {
            Integer count = cursor.getInt(0);
            subscriber.onNext(count);
        } else {
            subscriber.onNext(0);
        }
        return cursor;
    }
    private ContentObserver observer = new ContentObserver(new Handler()) {
        @Override
        public boolean deliverSelfNotifications() {
            return false;
        }
        @Override
        public void onChange(boolean selfChange) {
            Timber.d("New sms data changed");
            queryUnread(subscriber);
        }
    };
}

注1上述代码的问题是，由于registerObserver，它不能与.subscribeOn(Schedulers.io()一起调用，如果它被称为mainThread，那么查询也在它们上运行)

注意:将所有内容封装在单个Observable中是关键要求，也是这个问题的动机

我现在最好的想法是为活动创建一个HandlerThread，在那里我使用Observable并使用该线程的循环器。但是想知道是否有更好的替代方案，以及使用通用调度器(例如looperIoScheduler())是否有意义可能会导致问题。

Answer 1

在Observable链中，您可以随心所欲地更改线程。请看这里

函数rx.Observable#observeOn(rx.Scheduler)可以在链内的任何位置。尝试这样做(伪代码):

Observable.just(cursor)
        .observeOn(AndroidSchedulers.mainThread())
        .map((Cursor) -> {
                cursor.registerContentObserver(observer);
                return cursor;
            }
        }).observeOn(Schedulers.io());
subscriber.add(Subscriptions.create(() -> {
        cursor.unregisterContentObserver(observer);
        cursor.close();
    }));

Answer 2

好吧，那为什么不行呢?

public Observable<Integer> unreadCountObservable() {
    return Observable.defer(() -> Observable.create(subscriber -> {
        new UnreadCountObservable(subscriber);
    }))
    .subscribeOn(Schedulers.io());
}