RxJava 1中如何在Observable订阅回调前执行线程安全的一次性操作？

首先得明确你遇到的核心问题：既要保证更新操作只执行一次，又要确保它在所有订阅者的回调执行前完成。之前用doOnTerminate/doOnCompleted踩坑，大概率是两个原因：要么你的Observable是冷Observable（每个订阅者都会触发一次数据源执行，导致更新重复执行），要么线程调度的差异让你误以为顺序乱了（比如observeOn切换线程后，订阅者回调在另一个线程排队，但其实更新已经完成了，只是日志打印顺序误导了你）。

话不多说，直接上最优方案：

步骤1：将冷Observable转为热Observable（关键！） #

如果你的原始Observable是冷的（比如从网络/数据库拉取数据的Observable，每个订阅者订阅都会重新执行一次数据源逻辑），第一步必须把它转成热Observable，确保数据源只执行一次，更新操作也只会触发一次。

用publish().autoConnect()是最稳妥的方式：

// 你的原始冷Observable
Observable<YourData> originalColdObservable = ...;

// 转为热Observable，确保数据源只执行一次
Observable<YourData> hotObservable = originalColdObservable.publish().autoConnect();

publish()会把Observable转换成一个ConnectableObservable，autoConnect()会在第一个订阅者订阅时自动触发连接，后续订阅者共享同一个事件流。

步骤2：用lift操作符控制执行顺序 #

接下来用lift()自定义一个Operator，在事件转发给订阅者之前执行AtomicBoolean的更新。这种方式能绝对保证更新操作在所有订阅者的回调前完成，不受线程调度影响。

同时用另一个AtomicBoolean标记更新是否已执行，确保操作只跑一次：

AtomicBoolean targetFlag = new AtomicBoolean(false);
AtomicBoolean updateExecuted = new AtomicBoolean(false);

Observable<YourData> finalObservable = hotObservable.lift(
    new Observable.Operator<YourData, YourData>() {
        @Override
        public Subscriber<? super YourData> call(Subscriber<? super YourData> downstream) {
            return new Subscriber<YourData>(downstream) {
                @Override
                public void onNext(YourData data) {
                    // 如果你需要在发射数据时执行更新，就放在这里
                    if (updateExecuted.compareAndSet(false, true)) {
                        targetFlag.set(true);
                    }
                    // 转发事件给订阅者
                    downstream.onNext(data);
                }

                @Override
                public void onCompleted() {
                    // 如果你需要在Observable完成时执行更新，就放在这里
                    // 注意：如果onNext已经触发过更新，这里的compareAndSet会失败，避免重复执行
                    if (updateExecuted.compareAndSet(false, true)) {
                        targetFlag.set(true);
                    }
                    downstream.onCompleted();
                }

                @Override
                public void onError(Throwable e) {
                    // 根据你的需求决定是否在错误时执行更新
                    downstream.onError(e);
                }
            };
        }
    }
);

为什么这个方案可靠？ #

1. 一次性保证：updateExecuted.compareAndSet(false, true)是原子操作，确保即使多线程并发触发，更新也只会执行一次。
2. 顺序保证：事件必须经过自定义Subscriber的处理才能到达下游订阅者，更新操作和事件转发是串行执行的——不管上游用了subscribeOn还是下游用了observeOn，更新一定在订阅者的onNext/onCompleted之前完成。
3. 线程安全：AtomicBoolean本身就是线程安全的，它的set操作基于volatile变量，一旦更新完成，所有线程都能立即看到最新值。

额外说明 #

如果你不需要处理冷Observable的多订阅重复执行问题（比如你的Observable本来就是热的），可以跳过步骤1，直接在原始Observable上使用lift操作符即可。

另外，之前用doOn系列的问题：如果你的Observable是热的，doOnNext/doOnCompleted其实也能保证顺序，但lift的方式更直观，能让你明确控制事件流的每个环节，避免因线程调度的日志打印顺序产生误解。

内容的提问来源于stack exchange，提问作者jakub.g