如何创建一个带有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));
    });
}

- Zeyad Gasser

---

请查看 HandlerThread。 - pskink

@pskink 谢谢，这对我有用。 - Zeyad Gasser

使用 Kotlin 协程不是更好吗？ - IgorGanapolsky

Answer 1

HandlerThread可以胜任这项工作。

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

- Zeyad Gasser

Answer 2

这是带有循环器的线程示例：

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
    }
}

- MehDi

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 能够接收自动更新。

从 Realm 复制仍会导致结果的急切评估，因此这不是处理大型数据集的有效方式。

- EpicPandaForce

---

我喜欢你的方法，但这仍然没有解决我何时关闭realm实例的问题。我在一个不暴露给活动生命周期的类中进行我的数据库操作。因此，我需要在每个操作中打开和关闭实例，这迫使我使用copyFromRealm。我想保持这种模式，但我也想要实时更新。我将在我的问题中更新代码。 - Zeyad Gasser