如何使用looper创建后台线程

Question

谁能分享一个带有 Looper 的后台线程的实现，我可以将它传递给 subscribeOn(AndroidScheduler.from(/backgroundThreadWithLooper/))。

我需要这个，因为我正在尝试实现一个 DBService 类，该类在后台运行其所有操作，同时仍然获取实时对象更新。因此，当我应用 addChangeListener 时，会引发异常：

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.

DBService 代码：

public Observable<List> getAll(Class clazz) {
    return Observable.defer(() -> {
        Realm realm = Realm.getDefaultInstance();
        return realm.where(clazz).findAll().asObservable()
                .map(o -> realm.copyFromRealm((RealmResults) o))
                .doOnUnsubscribe(() -> closeRealm(realm))
                .doOnTerminate(() -> closeRealm(realm));
    });
}

Answer 1

HandlerThread 完成这项工作。

HandlerThread handlerThread = new HandlerThread("backgroundThread");
if (!handlerThread.isAlive())
    handlerThread.start();
AndroidSchedulers.from(handlerThread.getLooper());

Answer 2

这是一个带有 Looper 的线程示例：

public class GameLoop extends Thread {

@Override
public void run() {
    super.run();
    Looper.prepare(); // at first write this line of code
    //do something
    Looper.loop(); //and at the end write this line
}

}

Answer 3

这似乎对我有用

public class MainScopeListener
        extends Fragment {
    Realm realm;

    HandlerThread handlerThread;

    Scheduler looperScheduler;

    Observable<Realm> realmObservable;

    Subscription realmSubscription;

    Handler handler = new Handler(Looper.getMainLooper());

    public MainScopeListener() {
        setRetainInstance(true);
        realm = Realm.getDefaultInstance();
        Injector.INSTANCE.initializeComponent(realm);
        handlerThread = new HandlerThread("REALM_LOOPER") {
            @Override
            protected void onLooperPrepared() {
                super.onLooperPrepared();
                Log.i(getName(), "ON LOOPER PREPARED");
                handler.post(() -> {
                    looperScheduler = AndroidSchedulers.from(handlerThread.getLooper());
                    realmObservable = Observable.create(new Observable.OnSubscribe<Realm>() {
                        @Override
                        public void call(Subscriber<? super Realm> subscriber) {
                            final Realm observableRealm = Realm.getDefaultInstance();
                            observableRealm.setAutoRefresh(true);
                            final RealmChangeListener<Realm> listener = realm1 -> {
                                if (!subscriber.isUnsubscribed()) {
                                    subscriber.onNext(observableRealm);
                                }
                            };
                            subscriber.add(Subscriptions.create(() -> {
                                observableRealm.removeChangeListener(listener);
                                observableRealm.setAutoRefresh(false);
                                observableRealm.close();
                            }));
                            observableRealm.addChangeListener(listener);

                            // Immediately call onNext with the current value, as due to Realm's auto-update, it will be the latest
                            // value.
                            subscriber.onNext(observableRealm);
                        }
                    });
                    realmSubscription = realmObservable.unsubscribeOn(looperScheduler).subscribeOn(looperScheduler).subscribe(realm12 -> {
                        Log.i("REALM SUBSCRIPTION", "An event occurred on background thread!");
                    });
                });
            }
        };
        handlerThread.start();
    }

    public void configureRealmHolder(MainActivity.RealmHolder realmHolder) {
        realmHolder.realm = this.realm;
    }

    @Override
    public void onDestroy() {
        if(realmSubscription != null && !realmSubscription.isUnsubscribed() ) {
            realmSubscription.unsubscribe();
        }
        handlerThread.quit();
        realm.close();
        super.onDestroy();
    }
}

和

@SuppressWarnings("NewApi")
private Subscription writePeriodic() {
    return Observable.interval(2000, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread()) //
            .takeWhile(aLong -> counter < DogNames.values().length) //
            .observeOn(Schedulers.io())
            .doOnNext(aLong -> {
                try(Realm bgRealm = Realm.getDefaultInstance()) {
                    bgRealm.executeTransaction(realm1 -> {
                        long currentIndex = realm1.where(Dog.class).max(Dog.Fields.ID.getField()).longValue();
                        Dog dog = new Dog();
                        dog.setId(currentIndex + 1);
                        dog.setName(DogNames.values()[((Long) dog.getId()).intValue() % DogNames.values().length].name());
                        dog = realm1.copyToRealmOrUpdate(dog);
                        Log.i(TAG, "Realm write successful [" + counter + "] :: [" + dog.getName() + "].");
                        counter++;
                    });
                }
            }).subscribe();
}

结果

01-21 00:58:51.672 2094-2127/com.zhuinden.rxrealm I/DogView: Realm write successful [1] :: [Munch].
01-21 00:58:51.672 2094-2115/com.zhuinden.rxrealm I/REALM SUBSCRIPTION: An event occurred on background thread!

因此 HandlerThread Realm 能够接收自动更新。

从领域复制仍然会导致对结果的急切评估，因此它不是处理大型数据集的有效方法。