HarmonyOS中RxJava后台任务基类如何切换至主线程？

在HarmonyOS中没有Android平台的AndroidSchedulers.mainThread()，需要基于HarmonyOS的线程模型实现主线程调度器，以下是具体解决方案：

一、自定义HarmonyOS主线程Scheduler #

HarmonyOS的UI主线程依赖EventRunner，可以基于EventRunner.getMainEventRunner()实现专属的RxJava调度器：

public class HarmonyMainThreadScheduler extends Scheduler {
    private final EventRunner mainEventRunner;

    private HarmonyMainThreadScheduler() {
        this.mainEventRunner = EventRunner.getMainEventRunner();
    }

    public static HarmonyMainThreadScheduler instance() {
        return Holder.INSTANCE;
    }

    @Override
    public Worker createWorker() {
        return new MainThreadWorker(mainEventRunner);
    }

    private static class MainThreadWorker extends Worker {
        private final EventRunner eventRunner;

        MainThreadWorker(EventRunner eventRunner) {
            this.eventRunner = eventRunner;
        }

        @Override
        public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
            if (run == null) throw new NullPointerException("run is null");
            if (unit == null) throw new NullPointerException("unit is null");

            long delayMillis = unit.toMillis(delay);
            Runnable scheduledRun = () -> {
                if (!isDisposed()) {
                    run.run();
                }
            };

            EventHandler handler = new EventHandler(eventRunner);
            handler.postTask(scheduledRun, delayMillis);

            return () -> handler.removeTask(scheduledRun);
        }
    }

    private static class Holder {
        static final HarmonyMainThreadScheduler INSTANCE = new HarmonyMainThreadScheduler();
    }
}

二、修改BaseUseCase代码 #

将observeOn的参数替换为自定义的主线程调度器即可：

public void execute(DisposableObserver<O> observer, I input) {
    Preconditions.checkNotNull(observer);
    final Observable<O> observable = observable(input)
            .subscribeOn(Schedulers.io())
            .observeOn(HarmonyMainThreadScheduler.instance()); // 替换为自定义调度器
    addDisposable(observable.subscribeWith(observer));
}

三、普通Java应用的替代方案 #

如果是普通Java SE应用，没有UI主线程概念，可根据场景选择：

• 若需要单线程执行回调：使用Schedulers.single()
• 若为JavaFX应用：基于Platform.runLater()实现对应的Scheduler
• 自定义调度器：基于应用自身的事件循环机制实现

内容的提问来源于stack exchange，提问作者Maros Zelo