RxJava中如何选择仅在订阅后才发射新项的Subject？

嘿，这个问题我之前在项目里也纠结过，刚好可以给你详细拆解一下～

先搞懂PublishSubject文档里的「除非采取措施阻止」是什么意思 #

PublishSubject本身是热Observable，也就是说它一旦创建，如果有代码调用它的onNext()（不管是上游Observable推送，还是你直接调用），它会立刻尝试把事件发射出去——如果这时候还没有观察者订阅，这些事件就直接丢失了。

文档里说的「采取措施阻止」，其实就是不让PublishSubject在有观察者订阅之前就开始接收/发射事件，常见的两种方式：

• 手动控制事件发射时机：最简单的方式，就是确保你在调用subscribe()之后，才让上游Observable开始发射事件，或者才手动调用Subject的onNext()。比如你可以把Subject的引用封装起来，只在有订阅后才暴露发射事件的入口。
• 用ConnectableObservable延迟启动：把PublishSubject转成ConnectableObservable，这样只有当你调用connect()方法之后，它才会开始接收上游的事件并发射给已订阅的观察者。示例代码：

// 创建PublishSubject并包装成ConnectableObservable
PublishSubject<String> subject = PublishSubject.create();
ConnectableObservable<String> connectable = subject.publish();

// 先订阅观察者
connectable.subscribe(item -> System.out.println("观察者1收到：" + item));
connectable.subscribe(item -> System.out.println("观察者2收到：" + item));

// 现在才开始允许事件发射
connectable.connect();

// 这时候调用onNext，观察者就能收到了
subject.onNext("新事件1");
subject.onNext("新事件2");

这样就完全避免了订阅前丢失事件的风险，因为connect之前的事件根本不会进入Subject的发射流程。

有没有更合适的Subject类型？ #

如果想要一个不需要手动控制，天然就只会在有观察者订阅之后才开始发射后续新事件，并且不会缓存之前事件的Subject，其实PublishSubject已经是最贴合的了，但如果你怕自己不小心在订阅前调用了onNext导致丢事件，可以考虑这两个方向：

1. 自定义安全的Subject包装类 #

如果你想要通用的、支持多观察者、且只有订阅后才发射新事件的Subject，可以自己封装一个：

public class SafePublishSubject<T> extends Subject<T> {
    private final PublishSubject<T> delegate = PublishSubject.create();
    private volatile boolean hasSubscribers = false;

    @Override
    public void onSubscribe(Disposable d) {
        delegate.onSubscribe(d);
    }

    @Override
    public void onNext(T t) {
        if (hasSubscribers) {
            delegate.onNext(t);
        }
        // 无订阅者时直接丢弃事件
    }

    @Override
    public void onError(Throwable e) {
        if (hasSubscribers) {
            delegate.onError(e);
        }
    }

    @Override
    public void onComplete() {
        if (hasSubscribers) {
            delegate.onComplete();
        }
    }

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        hasSubscribers = true;
        delegate.subscribe(observer);
    }

    @Override
    public boolean hasObservers() {
        return delegate.hasObservers();
    }

    @Override
    public boolean hasThrowable() {
        return delegate.hasThrowable();
    }

    @Override
    public boolean hasComplete() {
        return delegate.hasComplete();
    }

    @Override
    public Throwable getThrowable() {
        return delegate.getThrowable();
    }
}

这个包装类会在有第一个观察者订阅后才开始转发onNext事件，订阅前的事件直接丢弃，完美符合你「仅在调用subscribe后才开始发射新项」的需求，而且支持多观察者。

2. UnicastSubject（仅适用于单个观察者场景） #

UnicastSubject是专门为单个观察者设计的，它会缓存订阅前的所有事件，直到第一个（也是唯一一个）观察者订阅，然后把缓存的事件和后续的新事件都发给观察者。不过它的限制是只能有一个观察者，多订阅会报错，而且不符合你要的「仅发射新项」（它会发送订阅前的缓存），所以只适合特定场景。

总结 #

• 如果你的场景可以自己控制事件发射时机，直接用PublishSubject就够了，记得不要在订阅前调用onNext，或者用ConnectableObservable来延迟启动。
• 如果需要更安全的自动控制，自己封装一个SafePublishSubject是最灵活的方案。
• UnicastSubject只适合单个观察者的场景，且会发送订阅前的缓存，不符合你的核心需求。

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