java - 订阅后合并PublishSubject

Question

这是我想要实现的目标:

PublishSubject<Integer> subject = PublishSubject.create();
    subject.subscribe(new Consumer<Integer>() {
        @Override
        public void accept(@NonNull Integer integer) throws Exception {
            System.out.println("item = " + integer);
        }
    });
    subject.mergeWith(Observable.just(1, 2));
    subject.onNext(3);
    /* 
    expected: 
    item = 1
    item = 2
    item = 3
    but received :
    item = 3
    */

我知道我可以做这样的事情:

PublishSubject.merge(subject, Observable.just(1,2)).subscribe(new Consumer<Integer>() {
        @Override
        public void accept(@NonNull Integer integer) throws Exception {
            System.out.println("item = " + integer); // emits 1 2 3
        }
    });

但问题是用户已经订阅了该主题。 我找不到一种优雅的方式。

编辑: 由于主体既是观察者又是订阅者，你可以这样做:

final PublishSubject<Integer> subject = PublishSubject.create();
    subject.subscribe(new Consumer<Integer>() {
        @Override
        public void accept(@NonNull Integer integer) throws Exception {
            System.out.println("item = " + integer);
        }
    });
    Observable.just(1,2).subscribe(subject);
    subject.onNext(3);//subscription
    /*
    expected:
    item = 1
    item = 2
    item = 3
    but received :
    item = 1
    item = 2
    */

Answer 1

将代码改为

    Subject<Integer> subject = PublishSubject.create();
    subject.subscribe(new Consumer<Integer>() {
        @Override
        public void accept(@NonNull Integer integer) throws Exception {
            System.out.println("item = " + integer);
        }
    });
    Observable.just(1,2).doOnNext(new Consumer<Integer>() {
        @Override
        public void accept(@NonNull Integer e) throws Exception {
            subject.onNext(e);
        }
    }).subscribe();

    subject.onNext(3);

但仍然如此。您需要订阅此 Observable。