RxJava2中的条件过滤器

Observable.just(1, 2, 3, 4, 5)
    .filter(item -> item % 2 == 0) // 只保留偶数
    .subscribe(System.out::println);
// 输出: 2, 4

Observable.range(1, 10)
    .take(3) // 只取前3个
    .subscribe(System.out::println);
// 输出: 1, 2, 3

Observable.range(1, 5)
    .takeLast(2) // 取最后2个
    .subscribe(System.out::println);
// 输出: 4, 5

Observable.range(1, 10)
    .takeWhile(i -> i < 5) // 当i<5时继续发出
    .subscribe(System.out::println);
// 输出: 1, 2, 3, 4

Observable.intervalRange(1, 10, 0, 100, TimeUnit.MILLISECONDS)
    .takeUntil(observable -> {
        // 3秒后发出数据，停止原Observable
        Observable.timer(3, TimeUnit.SECONDS).subscribe();
        return Observable.timer(3, TimeUnit.SECONDS);
    })
    .blockingSubscribe(System.out::println);
// 输出前30个数字（约3秒）

Observable.range(1, 5)
    .skip(2) // 跳过前2个
    .subscribe(System.out::println);
// 输出: 3, 4, 5

Observable.range(1, 5)
    .skipLast(2) // 跳过最后2个
    .subscribe(System.out::println);
// 输出: 1, 2, 3

Observable.range(1, 5)
    .skipWhile(i -> i < 3) // 跳过i<3的元素
    .subscribe(System.out::println);
// 输出: 3, 4, 5

Observable.intervalRange(1, 5, 0, 100, TimeUnit.MILLISECONDS)
    .skipUntil(Observable.timer(300, TimeUnit.MILLISECONDS))
    .blockingSubscribe(System.out::println);
// 跳过前3个，输出后2个

Observable.just(1, 2, 2, 3, 4, 4, 5)
    .distinct()
    .subscribe(System.out::println);
// 输出: 1, 2, 3, 4, 5

Observable.just(1, 2, 2, 3, 3, 3, 4, 5, 5)
    .distinctUntilChanged()
    .subscribe(System.out::println);
// 输出: 1, 2, 3, 4, 5

Observable.just(10, 20, 30, 40)
    .elementAt(2) // 获取第3个元素
    .subscribe(System.out::println);
// 输出: 30

Observable.just(5, 3, 8, 1)
    .first() // 获取第一个元素
    .subscribe(System.out::println);
// 输出: 5

Observable.just(5, 3, 8, 1)
    .last() // 获取最后一个元素
    .subscribe(System.out::println);
// 输出: 1

Observable.just(42)
    .single()
    .subscribe(System.out::println);

Observable.just(1, 2)
    .single() // 抛出异常，因为有多个元素
    .subscribe(System.out::println, Throwable::printStackTrace);

Observable.intervalRange(1, 10, 0, 100, TimeUnit.MILLISECONDS)
    .takeWhile(i -> i < 5) // 先取i<5
    .takeUntil(i -> i == 3) // 再取到i==3时停止
    .blockingSubscribe(System.out::println);
// 输出: 0, 1, 2, 3