用RxJava生成无限自然数序列

Question

我试图用RxJava编写一个简单的程序来生成无限自然数序列。到目前为止，我已经找到了使用observable.timer（）和observable.interval（）生成数字序列的两种方法。我不确定这些函数是否是解决这个问题的正确方法。我期待一个简单的函数，就像Java8中的函数一样，可以生成无限自然数。

import rx.Observable;
import rx.functions.Action1;

import java.util.stream.IntStream;

public class NaturalNumbers {

    public static void main(String[] args) {
        Observable<Integer> naturalNumbers = Observable.<Integer>create(subscriber -> {
            IntStream stream = IntStream.iterate(1, val -> val + 1);
            stream.forEach(naturalNumber -> subscriber.onNext(naturalNumber));
        });

        Action1<Integer> first = naturalNumber -> System.out.println("First got " + naturalNumber);
        Action1<Integer> second = naturalNumber -> System.out.println("Second got " + naturalNumber);
        Action1<Integer> third = naturalNumber -> System.out.println("Third got " + naturalNumber);
        naturalNumbers.subscribe(first);
        naturalNumbers.subscribe(second);
        naturalNumbers.subscribe(third);

    }
}

Answer 1

问题是您实现的onnaturalnumbers.subscribe（first）；、onsubscribe被调用，并且您正在对无限流执行foreach，因此程序从未终止。

处理它的一种方法是在不同的线程上异步订阅它们。为了方便地看到结果，我必须在流处理中引入睡眠：

Observable<Integer> naturalNumbers = Observable.<Integer>create(subscriber -> {
    IntStream stream = IntStream.iterate(1, i -> i + 1);
    stream.peek(i -> {
        try {
            // Added to visibly see printing
            Thread.sleep(50);
        } catch (InterruptedException e) {
        }
    }).forEach(subscriber::onNext);
});

final Subscription subscribe1 = naturalNumbers
    .subscribeOn(Schedulers.newThread())
    .subscribe(first);
final Subscription subscribe2 = naturalNumbers
    .subscribeOn(Schedulers.newThread())
    .subscribe(second);
final Subscription subscribe3 = naturalNumbers
    .subscribeOn(Schedulers.newThread())
    .subscribe(third);

Thread.sleep(1000);

System.out.println("Unsubscribing");
subscribe1.unsubscribe();
subscribe2.unsubscribe();
subscribe3.unsubscribe();
Thread.sleep(1000);
System.out.println("Stopping");