RXJAVA-FlatMap

Posted 2022-11-24 征服.刘华强

FlatMap将一个发送事件的上游Observable变换为多个发送事件的Observables，然后将它们发射的事件合并后放进一个单独的Observable里.

 

package com.netty.demo.vertx;

import io.reactivex.*;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Slf4j
public class RxJavaTest 

    public static void main(String[] args) throws InterruptedException 

        Observable.create(new ObservableOnSubscribe<Integer>() 
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception 
                emitter.onNext(1);
                emitter.onNext(2);
                emitter.onNext(3);
            
        ).observeOn(Schedulers.newThread()).subscribeOn(Schedulers.io())
        .flatMap(new Function<Integer, ObservableSource<String>>() 
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception 
                final List<String> list = new ArrayList<>();
                for (int i = 0; i < 3; i++) 
                    list.add("I am value " + integer);
                
                log.info(Arrays.toString(list.toArray()));
                return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
            
        ).subscribe(new Consumer<String>() 
            @Override
            public void accept(String s) throws Exception 
                log.info(s);
            
        );

        Thread.sleep(300000000);
    

结果是无序的

2021-02-22 15:43:14 [RxNewThreadScheduler-1] INFO  c.n.d.v.RxJavaTest:apply - [I am value 1, I am value 1, I am value 1]
2021-02-22 15:43:14 [RxNewThreadScheduler-1] INFO  c.n.d.v.RxJavaTest:apply - [I am value 2, I am value 2, I am value 2]
2021-02-22 15:43:14 [RxNewThreadScheduler-1] INFO  c.n.d.v.RxJavaTest:apply - [I am value 3, I am value 3, I am value 3]
2021-02-22 15:43:14 [RxComputationThreadPool-1] INFO  c.n.d.v.RxJavaTest:accept - I am value 1
2021-02-22 15:43:14 [RxComputationThreadPool-1] INFO  c.n.d.v.RxJavaTest:accept - I am value 1
2021-02-22 15:43:14 [RxComputationThreadPool-1] INFO  c.n.d.v.RxJavaTest:accept - I am value 1
2021-02-22 15:43:14 [RxComputationThreadPool-3] INFO  c.n.d.v.RxJavaTest:accept - I am value 3
2021-02-22 15:43:14 [RxComputationThreadPool-3] INFO  c.n.d.v.RxJavaTest:accept - I am value 2
2021-02-22 15:43:14 [RxComputationThreadPool-3] INFO  c.n.d.v.RxJavaTest:accept - I am value 2
2021-02-22 15:43:14 [RxComputationThreadPool-3] INFO  c.n.d.v.RxJavaTest:accept - I am value 2
2021-02-22 15:43:14 [RxComputationThreadPool-3] INFO  c.n.d.v.RxJavaTest:accept - I am value 3
2021-02-22 15:43:14 [RxComputationThreadPool-3] INFO  c.n.d.v.RxJavaTest:accept - I am value 3