RxJava主题不重复使用.repeat()

Question

谁能给我解释一下这种行为?

我正在学习RxJava，我想把我自己的事件发布到一个可观察链上。但是我在理解受试者的行为方面有一些问题。

这个代码:

PublishSubject<String> subject = PublishSubject.create();
subject.repeat(3)
       .subscribe(s -> System.out.println("subject emitted %s", s));
subject.onNext("one");
subject.onNext("two");
subject.onNext("three");
subject.onNext("four");
subject.onCompleted();
Observable.just("one", "two", "three", "four")
          .repeat(3)
          .subscribe(s -> System.out.println("observer emitted %s", s));

输出:

subject emitted one
subject emitted two
subject emitted three
subject emitted four
observer emitted one
observer emitted two
observer emitted three
observer emitted four
observer emitted one
observer emitted two
observer emitted three
observer emitted four
observer emitted one
observer emitted two
observer emitted three
observer emitted four

为什么在主题上忽略了。repeat(3) ?

谢谢!

Answer 1

在第一个subscribe之后，repeat将不会订阅PublicSubject，直到它收到onCompleted。您可以使用doOnSubscribe和doOnCompleted来调试它。如

    PublishSubject<String> subject = PublishSubject.create();
    subject.doOnSubscribe(() -> System.out.println("subject subscribe"))
            .doOnCompleted(() -> System.out.println("subject onCompleted"))
            .repeat(3)
            .subscribe(s -> System.out.printf("subject emitted %sn", s));
    subject.onNext("one");
    subject.onNext("two");
    subject.onNext("three");
    subject.onNext("four");
    subject.onCompleted();

执行顺序如下:

subscribe to PublicSubject
subject.onNext("one") => output "one"
subject.onNext("two") => output "two"
subject.onNext("three") => output "three"
subject.onNext("four") => output "four"
subject.onCompleted() => 
            output "onCompleted"
            subscribe to PublicSubject // a new Observer is added to PublicSubject and PublicSubject is still emitting `onCompleted`, 
                                       // so this new Observer will receive `onCompleted` at once
            output "onCompleted"
            subscribe to PublicSubject // a new Observer is added to PublicSubject and PublicSubject is still emitting `onCompleted`, 
                                       // so this new Observer will receive `onCompleted` at once
            output "onCompleted"

实际上，如果你想输出所有项目3次，你可以使用ReplaySubject

Answer 2

也许你可以这样使用.asObservable():

subject.asObservable().repeat(3)
       .subscribe(s -> System.out.println("subject emitted %s", s));