devbox
你好赵伟
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

RxJava（六）变换操作符_the method flatmapiterable(function

作者：你好赵伟 | 2024-03-01 22:55:46

赞

踩

the method flatmapiterable(function

变换操作符

  1. private void RxJava1() {
  2.  
  3.         ObservableSource<T> 另一个Observable;
  4.         Predicate<T> 条件;
  5.         prefetch 预取量;
  6.         maxConcurrency 最大并发数;
  7.  
  8.         //observeOn ： 指定观察者观察Observable的调度器
  9.         //subscribeOn ： 指定Observable执行任务的调度器
  10.  
  11.         //判断发射的所有数据项是否都满足Predicate条件
  12.         Observable.range(1,10).all(Predicate<T>);
  13.         //两个Observable中那个先发射了数据，就发射他的全部数据
  14.         Observable.range(1,10).ambWith(ObservableSource<T>);
  15.  
  16.         Observable.range(1,10).any(Predicate<T>);
  17.  
  18.         Observable.range(1,10).as(ObservableConverter<T,R>);
  19.  
  20.  
  21.         //定时定数收集Observable发射的数据放进一个集合，然后发射这个集合（包含收集的数据项）
  22.         Observable.range(1,10).buffer(int count);
  23.         Observable.range(1,10).buffer(int count,int skip);
  24.         Observable.range(1,10).buffer(long timeSpan,TimeUnit unit);
  25.         Observable.range(1,10).buffer(long timeSpan,TimeUnit unit,int count);
  26.         Observable.range(1,10).buffer(long timeSpan,long timeSkip,TimeUnit unit);
  27.  
  28.         //发射第一项，否则抛出异常NoSuchElementException
  29.         Flowable.just(1,2,3).blockingFirst(Integer defaultItem);
  30.         //以阻塞方式消耗发射的数据
  31.         Flowable.just(1,2,3).blockingForEach(new Consumer<Integer>);
  32.         //将此Flowable转换为Iterable
  33.         Flowable.just(1,2,3).blockingIterable();
  34.         Flowable.just(1,2,3).blockingIterable(int bufferSize);
  35.         //发射最后一项，否则抛出异常NoSuchElementException
  36.         Flowable.just(1,2,3).blockingLast();
  37.         //发射最后一项，若无数据则发射默认值
  38.         Flowable.just(1,2,3).blockingLast(Integer defaultItem);
  39.         //返回一个Iterable，它将会阻塞，直到此Flowable发出的新项目，并将其返回。
  40.         Flowable.just(1,2,3).blockingLatest();
  41.         //返回一个Iterable，在每次迭代时阻塞，直到此Flowable发出一个新项，然后Iterable返回该项
  42.         Flowable.just(1,2,3).blockingNext();
  43.         //发射单个值，如果它发出多个项目，则抛出IllegalArgumentException
  44.         Flowable.just(1,2,3).blockingSingle();
  45.         //发射单个值，如果它发出多个项目，则抛出IllegalArgumentException，若无数据发射，则返回默认值
  46.         Flowable.just(1,2,3).blockingSingle(Integer defaultItem);
  47.         //将Flowable运行至结束，忽略任何数据和异常
  48.         Flowable.just(1,2,3).blockingSubscribe(new Consumer<>);
  49.  
  50.         //立即执行
  51.         Observable.range(1,10).compose(ObservableTransformer<>);
  52.  
  53.         Observable.range(1,10).cacheWithInitialCapacity(int initialCapacity);
  54.  
  55.         //在发射之前强制将Observable发射的所有数据转换为指定类型
  56.         Observable.range(1,10).cast(Class<T>);
  57.  
  58.         //收集数据到一个集合，数组；T1为Callable发射数据，T2为Observable发射数据，分别组合被收集
  59.         Observable.range(1,10).collect(Callable<U> initialValueSuplier,BiConsumer<T1,T2>);
  60.         //收集数据到一个集合，数组,T1为U值，T2为Observable发射数据
  61.         Observable.range(1,10).collectInto(U,BiConsumer<T1,T2>);
  62.  
  63.  
  64.         //对原始Observable数据应用一个函数
  65.         Observable.range(1,10).map(Function<T,R>);
  66.         //对Observable数据子项进行一个函数变换，返回N个发射这些变换结果的Observables，然后组合这些Obsrvables
  67.         //发射的数据为ObservableSource<T>，当作自己的数据序列发射。
  68.         Observable.range(1,10).flatMap(Function<T,ObservableSource<T>>);
  69.         Observable.range(1,10).flatMap(Function<T,ObservableSource<T>>,boolean delayError,int maxCurrency);
  70.         Observable.range(1,10).flatMapIterable(Function<T,Iterable<T>>);
  71.         //类似FlatMap，只是严格按照顺序
  72.         Observable.range(1,10).concatMap(Function<T,ObservableSource<T>>,int prefetch);
  73.         Observable.range(1,10).concatMapDelayError(Function<T,ObservableSource<T>>);
  74.         //得到一个CompletableSource,
  75.         Observable.range(1,10).concatMapCompletable(Function<T, CompletableSource>);
  76.         Observable.range(1,10).concatMapCompletableDelayError(Function<T, CompletableSource>,boolean delayError);
  77.         Observable.range(1,10).concatMapEager(Function<T,ObservableSource<T>>);
  78.         Observable.range(1,10).concatMapEagerDelayError(Function<T,ObservableSource<T>>,boolean);
  79.         Observable.range(1,10).concatMapIterable(Function<T, Iterator<U>>);
  80.         Observable.range(1,10).concatMapIterable(Function<Integer, Iterator<U>>,int prefetch);
  81.         Observable.range(1,10).concatMapMaybe(Function<T, Maybe<T>>);
  82.         Observable.range(1,10).concatMapMaybeDelayError();
  83.         Observable.range(1,10).concatMapSingle(Function<T, Single<U>>);
  84.         Observable.range(1,10).concatMapSingleDelayError();
  85.         //
  86.         Observable.range(1,10).concatWith(CompletableSource);
  87.         Observable.range(1,10).concatWith(ObservableSource<T>);
  88.         Observable.range(1,10).concatWith(Maybe<T>);
  89.         Observable.range(1,10).concatWith(Single<T>);
  90.  
  91.         //在同一线程，结果同concatMap。当在不同线程，将Observable发射的数据变换为Observables集合，
  92.         // 当原始Observable发射一个新的数据（Observable）时，它将取消订阅前一个Observable
  93.         Observable.range(1,10).switchMap(Function<T,ObservableSource<T>>);
  94.  
  95.         //判断在发射的所有数据项中是否包含指定的数据
  96.         Observable.range(1,10).contains(Object element);
  97.  
  98.         //发射数据项的总数
  99.         Observable.range(1,10).count();
  100.  
  101.         //缓存Observable发射的数据序列并发射相同的数据序列给后续的订阅者
  102.         Observable.range(1,10).cache();
  103.  
  104.         //发射过快timeOut，丢弃前一次的数据；发射过慢timeOut，马上进行发射；用法同throttleWithTimeout
  105.         Observable.range(1,10).debounce(long timeOut,TimeUnit unit);
  106.         Observable.range(1,10).debounce(long timeOut,TimeUnit unit,Scheduler scheduler);
  107.         Observable.range(1,10).debounce(Function<T,ObservableSource<U>>);
  108.  
  109.         //如果原始Observable正常终止后仍然没有发射任何数据，就发射一个默认值,内部调用的switchIfEmpty。
  110.         Observable.range(1,10).defaultIfEmpty(Integer defaultItem);
  111.  
  112.         //延时发射Observable的结果
  113.         Observable.range(1,10).delay(long delay,TimeUnit unit);
  114.         Observable.range(1,10).delay(long delay,TimeUnit unit,boolean delayError);
  115.         Observable.range(1,10).delay(long delay,TimeUnit unit,Scheduler scheduler);
  116.         //延时处理订阅请求
  117.         Observable.range(1,10).delaySubscription(long delay,TimeUnit unit);
  118.         Observable.range(1,10).delaySubscription(long delay,TimeUnit unit,Scheduler scheduler);
  119.  
  120.         //过滤重复数据
  121.         Observable.range(1,10).distinct();
  122.         Observable.range(1,10).distinct(Function<T,Key>);
  123.         //过滤连续重复数据
  124.         Observable.range(1,10).distinctUntilChanged(Function<T,Key>);
  125.         Observable.range(1,10).distinctUntilChanged(BiPredicate<T1,T2>);
  126.  
  127.         //发射指定index项数据，没有则发射默认值
  128.         Observable.range(1,10).elementAt(long index);
  129.         Observable.range(1,10).elementAt(long index,Integer defaultItem);
  130.         Observable.range(1,10).elementAtOrError(long index);
  131.         Observable.range(1,10).firstElement();
  132.         Observable.range(1,10).lastElement();
  133.         //不关心发射的数据，只发射onError,onComplete
  134.         Observable.range(1,10).ignoreElements();
  135.  
  136.         Observable.range(1,10).first(Integer defaultItem);
  137.         Observable.range(1,10).firstOrError();
  138.         Observable.range(1,10).last(Integer defaultItem);
  139.         Observable.range(1,10).lastOrError();
  140.  
  141.         //BlockingObservable是一个阻塞的Observable。普通的Observable 转换为 BlockingObservable，可以使用
  142.         //Observable.toBlocking( )方法或者BlockingObservable.from( )方法。内部通过CountDownLatch实现了阻塞操作。
  143.         //对BlockingObservable发射的每一项数据调用一个方法，会阻塞直到Observable完成。
  144.         Observable.range(1,10).forEach(Consumer<T>);
  145.         Observable.range(1,10).forEachWhile(Predicate<T> onNext,onError,onComplete);
  146.  
  147.         //对每一项数据应用一个函数，按条件分配key，相同key分配到一个组，然后发射这些组
  148.         Observable.range(1,10).groupBy(Function<T,Key>);
  149.         Observable.range(1,10).groupBy(Function<T,Key>,boolean delayError);
  150.         Observable.range(1,10).groupBy(Function<T,Key>,Function<T,V>);
  151.         Observable.range(1,10).groupBy(Function<T,Key>,Function<T,V>,boolean delayError);
  152.  
  153.         //将主题封装在一个可观察的对象中，这使得消费代码无法将其转换回主题，是隐藏实现细节的一种防御机制。
  154.         Observable.range(1,10).hide();
  155.  
  156.         //用于判断Observable有没有发射数据。不为空false，如果只收到了onComplete通知则为true。
  157.         Observable.range(1,10).isEmpty();
  158.  
  159.         //两个Observable在交叉发射时间内，应用一个函数结合他们的数据，只要其中一个发射了onComplete则结束
  160.         //ObLeft.jion(ObRight,Func1,Func2,BiFunc)
  161.         Observable.range(1,10).join(ObservableSource<T> right,Function<T,
  162.         ObservableSource<T>> funcLeft, Function<T,ObservableSource<T>> funcRight,BiFunction<T1,T2,R> funcR);
  163.         //结果相同，只是BiFunction组合的是Observable
  164.         Observable.range(1,10).groupJoin(参数同上);
  165.  
  166.         //所有操作符的基础,内部都有调用
  167.         Observable.range(1,10).lift(ObservableOperator<R,T>);
  168.  
  169.         // 将Observable转换成一个通知列表,发送通知
  170.         Observable.range(1,10).materialize();
  171.         //将通知转换为一个Observable
  172.         Observable.range(1,10).dematerialize();
  173.  
  174.         //组合两个Observable，Maybe，Single，Complable
  175.         Observable.range(1,10).mergeWith(ObservableSource<T>);
  176.  
  177.         //过滤指定类型的数据，与filter类似
  178.         Observable.range(1,10).ofType(Class<T>);
  179.  
  180.         //当原始Observable在遇到错误时，使用备用Observable。可以处理所有的错误
  181.         Observable.range(1,10).onErrorResumeNext(ObservableSource<T> next);
  182.         Observable.range(1,10).onErrorResumeNext(Function<Throwable,ObservableSource<T>>);
  183.         //当原始Observable在遇到错误时发射一个特定的数据
  184.         Observable.range(1,10).onErrorReturn(Function<Throwable,T>);
  185.         Observable.range(1,10).onErrorReturnItem(item);
  186.         //当原始Observable在遇到错误时，使用备用Observable。只能处理异常。
  187.         Observable.range(1,10).onExceptionResumeNext(ObservableSource<T> next);
  188.  
  189.         //当数据切断的时候，要执行此操作，防止内存泄露
  190.         Observable.range(1,10).onTerminateDetach();
  191.  
  192.         //当接收到onComplete()会触发重新订阅，默认情况下运行在一个新的线程上
  193.         Observable.range(1,10).repeat();
  194.         Observable.range(1,10).repeat(long time);
  195.         //直到满足什么条件就重新订阅；BooleanSupplier存储一个boolean值，.getAsBoolean()获取boolean值；
  196.         Observable.range(1,10).repeatUntil(BooleanSupplier stop);
  197.         //当满足什么条件就会重新订阅
  198.         Observable.range(1,10).repeatWhen(Function<Observable<T>,ObservableSource<R>>);
  199.  
  200.         //对数据序列应用一个函数，然后发射结果
  201.         Observable.range(1,10).reduce(BiFunction<T1,T2,R>);
  202.  
  203.         //生成connectableObservable，使得无论什么时候订阅都能收到完整数据
  204.         Observable.range(1,10).replay();
  205.         //参数，1个，2个，3个组合
  206.         Observable.range(1,10).replay(int bufferSize,long Time,TimeUnit unit,Scheduler shceduler);
  207.         Observable.range(1,10).replay(Function<Observable<T>,ObservableSource<R>>);
  208.  
  209.         //当原始Observable在遇到错误时进行重试。
  210.         Observable.range(1,10).retry();
  211.         Observable.range(1,10).retry(long time);
  212.         Observable.range(1,10).retry(Predicate<Throwable>);
  213.         Observable.range(1,10).retry(long time,Predicate<Throwable>);
  214.         Observable.range(1,10).retry(BiPredicate<T,Throwable>);
  215.         Observable.range(1,10).retryUntil(BooleanSupplier stop);
  216.         //当原始Observable在遇到错误，将错误传递给另一个ObservableSource<T>来决定是否要重新订阅这个Observable,
  217.         Observable.range(1,10).retryWhen(Function<Observable<Throwable>,ObservableSource<T>>);
  218.  
  219.         //如果Observable终止时只发射了一个值，返回那个值，否则抛出异常或者发射默认值。
  220.         Observable.range(1,10).single(T defaultItem);
  221.         Observable.range(1,10).singleOrError();
  222.         Observable.range(1,10).singleElement();
  223.  
  224.         //定期发射Observable最近的数据(最后一个)
  225.         Observable.range(1,10).sample(long peroid,TimeUnit unit);
  226.         Observable.range(1,10).sample(long peroid,TimeUnit unit,boolean emitLast);
  227.         Observable.range(1,10).sample(ObservableSource<U>);
  228.         Observable.range(1,10).sample(ObservableSource<U>,boolean emitLast);
  229.         Observable.range(1,10).sample(long peroid,TimeUnit unit,Scheduler scheduler);
  230.         Observable.range(1,10).sample(long peroid,TimeUnit unit,Scheduler scheduler,boolean emitLast);
  231.  
  232.         //与reduce很像，对Observable发射的每一项数据应用一个函数，然后按顺序依次发射每一个值。
  233.         Observable.range(1,10).scan(BiFunction<T1,T2,R>);
  234.         Observable.range(1,10).scanWith(...);
  235.  
  236.         //跳过定时定量数据
  237.         Observable.range(1,10).skip(long count);
  238.         Observable.range(1,10).skip(long time,TimeUnit unit);
  239.         Observable.range(1,10).skip(long time,TimeUnit unit,Scheduler scheduler)
  240.         Observable.range(1,10).skipLast(同上);
  241.         Observable.range(1,10).skipLast(同上,boolean delayError);
  242.         //忽略原始Observable发射的数据，直到自定义的Observable发射了一个数据，Observer开始接收数据。
  243.         Observable.range(1,10).skipUntil(ObservableSource<T> other);
  244.         //忽略，直到指定条件不成立
  245.         Observable.range(1,10).skipWhile(Predicate<T>);
  246.         //发射定时定量数据
  247.         Observable.range(1,10).take(同skip);
  248.         Observable.range(1,10).takeLast(同skip );
  249.         Observable.range(1,10).takeUntil(ObservableSource<T> other);
  250.         Observable.range(1,10).takeUntil(Predicate<T>);
  251.         Observable.range(1,10).takeWhile(Predicate<T>);
  252.  
  253.         //共享数据，每个Observer都可已收到相同的数据
  254.         Observable.range(1,10).share();
  255.  
  256.         //强制Observable按次序发射数据并且要求功能是完好的
  257.         Observable.range(1,10).serialize();
  258.  
  259.         //排序
  260.         Observable.range(1,10).sorted();
  261.         //Comparator外比较器
  262.         Observable.range(1,10).sorted(Comparator<T>);
  263.  
  264.         //在发射数据序列之前添加数据序列
  265.         Observable.range(1,10).startWith(T item);
  266.         Observable.range(1,10).startWith(Iterable<T>);
  267.         Observable.range(1,10).startWith(ObservableSource<T> other);
  268.         Observable.range(1,10).startWithArray(Array[]);
  269.  
  270.         //如果原始Observable正常终止后仍然没有发射任何数据，就使用备用的Observable。
  271.         Observable.range(1,10).switchIfEmpty(ObservableSource<T>);
  272.  
  273.         //只发射第一个数据，忽略skipDuration内发射的数据
  274.         Observable.range(1,10).throttleFirst( long skipDuration, TimeUnit unit, Scheduler scheduler);
  275.         Observable.range(1,10).throttleFirst( long windowDuration, TimeUnit unit);
  276.         //定期发射Observable最近的数据(最后一个)
  277.         Observable.range(1,10).throttleLast(同上);
  278.         Observable.range(1,10).throttleLatest(long timeout,TimeUnit unit);
  279.         Observable.range(1,10).throttleLatest(long timeout,TimeUnit unit,boolean emitLast);
  280.         Observable.range(1,10).throttleLatest(long timeout,TimeUnit unit,Scheduler shceduler,boolean emitLast);
  281.         //发射过快timeOut，丢弃前一次的数据；发射过慢timeOut，马上进行发射
  282.         Observable.range(1,10).throttleWithTimeout(long timeout,TimeUnit unit);
  283.         Observable.range(1,10).throttleWithTimeout(long timeout,TimeUnit unit,Scheduler scheduler);
  284.  
  285.         //发射两个数据之间的间隔时间
  286.         Observable.range(1,10).timeInterval();
  287.         Observable.range(1,10).timeInterval(TimeUnit unit);
  288.         Observable.range(1,10).timeInterval(Scheduler scheduler);
  289.         Observable.range(1,10).timeInterval(TimeUnit unit,Scheduler scheduler);
  290.  
  291.         //给Observable发射的每个数据项添加一个时间戳。数据带上时间
  292.         Observable.range(1,10).timestamp(参数同上);
  293.  
  294.         //如果原始Observable过了指定的一段时长没有发射任何数据，就发射一个异常或者使用备用的Observable
  295.         Observable.range(1,10).timeout(long timeout,TimeUnit unit);
  296.         Observable.range(1,10).timeout(long timeout,TimeUnit unit,Scheduler shceduler);
  297.         Observable.range(1,10).timeout(long timeout,TimeUnit unit,ObservableSource<T> other);
  298.  
  299.         //转换
  300.         Observable.range(1,10).to(Function<Observable<T>,R>);
  301.         Observable.range(1,10).toFlowable(BackpressureStrategy);
  302.         Observable.range(1,10).toFuture();
  303.         //收集原始Observable发射的所有数据到一个List集合
  304.         Observable.range(1,10).toList();
  305.         //将数据序列转换为一个Map,根据数据生成Key，Value
  306.         Observable.range(1,10).toMap(Function<T,K>,Function<T,V>);
  307.         //将数据序列转换为一个Map,根据数据生成Key，Value,类似于toMap，不同的地方在于map的value是一个集合。
  308.         Observable.range(1,10).toMultimap(Function<T,K>,Function<T,V>, Collection<V>);
  309.  
  310.         //收集原始Observable发射的所有数据到一个有序List集合
  311.         Observable.range(1,10).toSortedList();
  312.  
  313.         //将原始Observerable拆分成包含n项的窗口，然后发射这些窗口；
  314.         Observable.range(1,10).window(long count);
  315.         Observable.range(1,10).window(long count,long skip);
  316.         Observable.range(1,10).window(long timeSpan,TimeUnit unit);
  317.         Observable.range(1,10).window(long count,long skip,int bufferSize);
  318.  
  319.         Observable.range(1,10).withLatestFrom(ObservableSource<T> other,BiFunction<T2,T2,R>);
  320.         Observable.range(1,10).withLatestFrom(ObservableSource<T>[] others,Function<T[],R>);
  321.         Observable.range(1,10).withLatestFrom(Iterable<ObservableSource<T>> iterables,Function<T[],R>);        //按顺序依次组合多个Observable的数据项应用一个函数，直到数据项最少的那个Observable用完数据项为止不再组合，然后发射组合项
  322.         Observable.range(1,10).zipWith(ObservableSource<T>,BiFunction<T1,T2,R>);
  323.         Observable.range(1,10).zipWith(ObservableSource<T>,BiFunction<T1,T2,R>,boolean delayError);
  324.         Observable.range(1,10).zipWith(ObservableSource<T>,BiFunction<T1,T2,R>,boolean delayError,int bufferSize);
  325.         Observable.range(1,10).zipWith(Iterable<T>,BiFunction<T1,T2,R>);
  326.     }


map    对原始Observable的数据进行一个函数操作。
flatMap    对原始Observable的数据进行一个函数操作，然后将每一项数据合并，作为自己的数据序列发射，合并的数据顺序可能交错。

  1. private void rxJavTest() {
  2.  
  3.        Observable.just(1,2,3)
  4.                .subscribeOn(Schedulers.io())
  5.                .flatMap(new Function<Integer, ObservableSource<Integer>>() {
  6.                    @Override
  7.                    public ObservableSource<Integer> apply(Integer integer) throws Exception {
  8.                        integer = integer * 2;
  9.                        return Observable.just(integer);
  10.                    }
  11.                }).observeOn(AndroidSchedulers.mainThread())
  12.                .subscribe(new Consumer<Integer>() {
  13.                    @Override
  14.                    public void accept(Integer integer) {
  15.                        Log.d(TAG, "accept: " + integer);
  16.                    }
  17.                });
  18.     }


输出结果：
accept: 2
accept: 4
accept: 6

concatMap    同flatMap，只是合并后数据严格按照顺序发射。
switchMap    在同一线程，结果同concatMap。当在不同线程，如newThread则只发射最近请求。

  1.  private void rxJavTest() {
  2.  
  3.        Observable.just(1,2,3)
  4.                .subscribeOn(Schedulers.io())
  5.                .switchMap(new Function<Integer, ObservableSource<Integer>>() {
  6.                    @Override
  7.                    public ObservableSource<Integer> apply(Integer integer) throws Exception {
  8.                        integer = integer * 2;
  9.                        return Observable.just(integer).subscribeOn(Schedulers.newThread());
  10.                    }
  11.                }).observeOn(AndroidSchedulers.mainThread())
  12.                .subscribe(new Consumer<Integer>() {
  13.                    @Override
  14.                    public void accept(Integer integer) {
  15.                        Log.d(TAG, "accept: " + integer);
  16.                    }
  17.                });
  18.     }


输出结果：
accept: 6

groupBy 对每一项数据应用一个函数，按条件分配key，相同key分配到一个组，然后发射这些组。

  1.  private void rxJavTest() {
  2.         
  3.        Observable.range(1,15)
  4.                .subscribeOn(Schedulers.io())
  5.                .groupBy(new Function<Integer, Integer>() {
  6.                    @Override
  7.                    public Integer apply(Integer integer) {
  8.                        //分为5组
  9.                        return integer % 5;
  10.                    }
  11.                }).observeOn(AndroidSchedulers.mainThread())
  12.                .subscribe(new Consumer<GroupedObservable<Integer, Integer>>() {
  13.                    @Override
  14.                    public void accept(GroupedObservable<Integer, Integer> groupedObservable) throws Exception {
  15.                        Log.d(TAG, "accept: group key = " + groupedObservable.getKey());
  16.                        groupedObservable.subscribe(new Consumer<Integer>() {
  17.                            @Override
  18.                            public void accept(Integer integer) throws Exception {
  19.                                Log.d(TAG, "accept: group 数据项 ：" + integer);
  20.                            }
  21.                        });
  22.                    }
  23.                });
  24.     }



输出结果：
accept: group key = 1
accept: group 数据项 ：1
accept: group 数据项 ：6
accept: group 数据项 ：11
accept: group key = 2
accept: group 数据项 ：2
accept: group 数据项 ：7
accept: group 数据项 ：12
accept: group key = 3
accept: group 数据项 ：3
accept: group 数据项 ：8
accept: group 数据项 ：13
accept: group key = 4
accept: group 数据项 ：4
accept: group 数据项 ：9
accept: group 数据项 ：14
accept: group key = 0
accept: group 数据项 ：5
accept: group 数据项 ：10
accept: group 数据项 ：15buffer    定期收集原始Observable发射的数据形成一个包裹，然后当成自己的数据序列发射这个包裹。

  1. private void rxJavTest() {
  2.  
  3.         Observable.range(1, 10)
  4.                 .subscribeOn(Schedulers.io())
  5.                 .buffer(5)
  6.                 .observeOn(AndroidSchedulers.mainThread())
  7.                 .subscribe(new Consumer<List<Integer>>() {
  8.                     @Override
  9.                     public void accept(List<Integer> integers) throws Exception {
  10.                         for (int i = 0; i < integers.size(); i++) {
  11.                             Log.d(TAG, "list大小" + integers.size()
  12.                                     +"   list包裹中的数据项：" + integers.get(i));
  13.                         }
  14.                     }
  15.                 });
  16.     }


输出结果：
list大小5 list包裹中的数据项：1
list大小5 list包裹中的数据项：2
list大小5 list包裹中的数据项：3
list大小5 list包裹中的数据项：4
list大小5 list包裹中的数据项：5
list大小5 list包裹中的数据项：6
list大小5 list包裹中的数据项：7
list大小5 list包裹中的数据项：8
list大小5 list包裹中的数据项：9
list大小5 list包裹中的数据项：10

bufer(n,skip)    Observer每接受到一个原始Observable发射的数据，就用n项数据填充缓存：接受到数据项 --> 之后n-1项 ；然后发射这些缓存。

  1. private void rxJavTest() {
  2.  
  3.         Observable.range(1, 10)
  4.                 .subscribeOn(Schedulers.io())
  5.                 .buffer(5,1)
  6.                 .observeOn(AndroidSchedulers.mainThread())
  7.                 .subscribe(new Consumer<List<Integer>>() {
  8.                     @Override
  9.                     public void accept(List<Integer> integers) throws Exception {
  10.                         Log.d(TAG, "accept: List :" + integers.size());
  11.                         for (int i = 0; i < integers.size(); i++) {
  12.                             Log.d(TAG, "accept: 数据项：" + integers.get(i));
  13.                         }
  14.                     }
  15.                 });
  16.     }



输出结果：
accept: List :5    [1,2,3,4,5]
accept: List :5    [2,3,4,5,6]
accept: List :5    [3,4,5,6,7]
accept: List :5    [4,5,6,7,8]
accept: List :5    [5,6,7,8,9]
accept: List :5    [6,7,8,9,10]
accept: List :4    [7,8,9,10]
accept: List :3    [8,9,10]
accept: List :2    [9,10]
accept: List :1    [10]

window    将原始Observerable拆分成包含n项的窗口Observable，然后发射这些窗口；
private void rxJavTest() {

  1.         Observable.range(1, 10)
  2.                 .subscribeOn(Schedulers.io())
  3.                 .window(5)
  4.                 .observeOn(AndroidSchedulers.mainThread())
  5.                 .subscribe(new Consumer<Observable<Integer>>() {
  6.                     @Override
  7.                     public void accept(Observable<Integer> integerObservable) throws Exception {
  8.                         Log.d(TAG, "accept: 一个Window包含：");
  9.                         integerObservable.subscribe(new Consumer<Integer>() {
  10.                             @Override
  11.                             public void accept(Integer integer) throws Exception {
  12.                                 Log.d(TAG, "accept: 数据项：" + integer);
  13.                             }
  14.                         });
  15.                     }
  16.                 });
  17.     }



输出结果：
accept: 一个Window包含：
accept: 数据项：1
accept: 数据项：2
accept: 数据项：3
accept: 数据项：4
accept: 数据项：5
accept: 一个Window包含：
accept: 数据项：6
accept: 数据项：7
accept: 数据项：8
accept: 数据项：9
accept: 数据项：10

repeat 创建一个重复发射原始数据序列的Observable。
（1）repeat(n) 重复发射n次。
（2）repeatWhen    当满足某种条件时才重新订阅和发送原始数据。

  1.         final long start = System.currentTimeMillis();
  2.         Observable.just(1,2,3)
  3.                 .repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
  4.                     @Override
  5.                     public ObservableSource<?> apply(Observable<Object> objectObservable) throws Exception {
  6.                         return Observable.timer(3,TimeUnit.SECONDS);
  7.                     }
  8.                 }).subscribe(new Consumer<Integer>() {
  9.             @Override
  10.             public void accept(Integer integer) throws Exception {
  11.                 Log.d(TAG, "值 ：" + integer + " 收到时间 ：" + (System.currentTimeMillis() - start));
  12.             }
  13.         });


输出结果：
值 ：1 收到时间 ：22
值 ：2 收到时间 ：22
值 ：3 收到时间 ：22
值 ：1 收到时间 ：3023
值 ：2 收到时间 ：3024
值 ：3 收到时间 ：3024

（3）repeatUntil 直到某个条件就不再重复发射原始数据。

  1. final long start = System.currentTimeMillis();
  2.         Observable.just(1,2,3)
  3.                 .repeatUntil(new BooleanSupplier() {
  4.                     @Override
  5.                     public boolean getAsBoolean() throws Exception {
  6.                         //重复发射数据，直到时间走过50毫秒
  7.                         return (System.currentTimeMillis() - start) > 50;
  8.                     }
  9.                 }).subscribe(new Consumer<Integer>() {
  10.             @Override
  11.             public void accept(Integer integer) throws Exception {
  12.                 Log.d(TAG, "值 ：" + integer + " 收到时间 ：" + (System.currentTimeMillis() - start));
  13.             }
  14.         });



打印结果太多，就不贴出了。
 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/你好赵伟/article/detail/177374
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。