rxjava中操作符怎么用

RxJava中的操作符非常丰富，它们可以帮助你处理各种数据流和转换。以下是一些常用的RxJava操作符及其用法：

1. 创建操作符

• just(T... values): 创建一个发射指定值的Observable。

  Observable.just("Hello", "World")
      .subscribe(System.out::println);
  

• fromArray(T[] array): 将数组转换为Observable。

  Observable.fromArray(new String[]{"A", "B", "C"})
      .subscribe(System.out::println);
  

• interval(long initialDelay, long period, TimeUnit unit): 创建一个定时发射递增数字的Observable。

  Observable.interval(1, TimeUnit.SECONDS)
      .take(5) // 只取前5个值
      .subscribe(System.out::println);
  

2. 转换操作符

• map(Function<? super T, ? extends R> mapper): 对Observable发射的每个数据项应用一个函数，并返回一个新的Observable。

  Observable.just("1", "2", "3")
      .map(Integer::parseInt)
      .subscribe(System.out::println); // 输出: 1, 2, 3
  

• flatMap(Function<? super T, ? extends ObservableSource<R>> mapper): 将Observable发射的每个数据项转换为另一个Observable，并将这些Observable合并成一个单一的Observable。

  Observable.just("1", "2", "3")
      .flatMap(s -> Observable.just(s.split("")))
      .subscribe(System.out::println); // 输出: 1, 2, 3
  

3. 过滤操作符

• filter(Predicate<? super T> predicate): 只发射满足指定条件的数据项。

  Observable.range(1, 10)
      .filter(n -> n % 2 == 0) // 只取偶数
      .subscribe(System.out::println); // 输出: 2, 4, 6, 8, 10
  

• take(int count): 只取前N个数据项。

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

4. 组合操作符

• zip(Observable<? extends T> other, BiFunction<? super T, ? super T, ? extends R> zipper): 将两个Observable的数据项按顺序配对，并应用一个函数生成新的数据项。

  Observable.just(1, 2, 3)
      .zip(Observable.just("a", "b", "c"), (n, s) -> n + s)
      .subscribe(System.out::println); // 输出: 1a, 2b, 3c
  

• concatWith(Observable<? extends T> other): 将另一个Observable的数据项追加到当前Observable的末尾。

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

5. 错误处理操作符

• onErrorResumeNext(Function<Throwable, ? extends ObservableSource<? extends T>> function): 当发生错误时，使用指定的函数生成一个新的Observable来继续发射数据。

  Observable.just(1, 2, 3)
      .map(n -> {
          if (n == 2) throw new RuntimeException("Error on 2");
          return n;
      })
      .onErrorResumeNext(throwable -> Observable.just(-1)) // 发生错误时发射-1
      .subscribe(System.out::println); // 输出: 1, -1, 3
  

6. 订阅操作符

• subscribe(Observer<? super T> observer): 订阅Observable并开始接收数据。

  Observable.just("Hello", "World")
      .subscribe(new Observer<String>() {
          @Override
          public void onSubscribe(Disposable d) {
              System.out.println("Subscribed");
          }
  
          @Override
          public void onNext(String s) {
              System.out.println(s);
          }
  
          @Override
          public void onError(Throwable e) {
              System.out.println("Error: " + e.getMessage());
          }
  
          @Override
          public void onComplete() {
              System.out.println("Completed");
          }
      });
  

这些只是RxJava中众多操作符的一部分。为了更好地理解和使用这些操作符，建议查阅RxJava的官方文档和示例代码。