Spring Reactor map与flatMap操作符详解

傷城~ 2022-12-25 10:54 202阅读 0赞

上一篇：[Spring Reactor 操作符详解][Spring Reactor] 中讲了创建类与时间类操作符；  
[第三篇：Spring Reactor parallel并发与线程切换 实战][Spring Reactor parallel_]

## 转换与组合操作符 ##

#### map 操作符 ####

// 模仿数据库查询
     List<UserPO> userList = new ArrayList<>();
     UserPO userPO = new UserPO();
     userPO.setName("张三");
     userPO.setMobilePhone("1321111111");
     userList.add(userPO);
     // 对查询出来的数据进行逻辑处理
     List<UserVO> block = Flux.fromIterable(userList).map(new Function<UserPO, UserVO>() {
         @Override
         public UserVO apply(UserPO userPO) {
             UserVO userVO = new UserVO();
             userVO.setName(userPO.getName());
             String mobilePhone = userPO.getMobilePhone();
             StringBuilder sb = new StringBuilder();
             // 电话号码隐藏处理
             sb.append(mobilePhone, 0, 3).append("***").append(mobilePhone.substring(mobilePhone.length() - 3));
             userVO.setPhone(sb.toString());
             return userVO;
         }
     }).collectList().block();
     System.out.println(block.toString());
     // 输出结果
     [UserVO(name=张三, phone=132***111)]

**map操作符示意图：**  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoYW5nZTk4NzY1NDMyMQ_size_16_color_FFFFFF_t_70][]

*  map 操作符的作用就是将一种对象类型转换为另一种对象类型；
    
     *  第一个参数：源事件类型（UserPO）
     *  第二个参数：目标事件类型（UserVO）,可以是List,Map等各种Object类型；
 *  通过collectList进行收集，这里和jdk8的Stream收集返回类似；

### FlatMap操作符 ###

现在有个奇怪的需求：统计手机号码中有多少位不重复的数字

// 模仿数据库查询
     List<UserPO> userList = new ArrayList<>();
      UserPO userPO = new UserPO();
      userPO.setName("张三");
      userPO.setMobilePhone("1321111111");
      userList.add(userPO);
      // 对查询出来的数据进行逻辑处理
      Flux.fromIterable(userList).flatMap(new Function<UserPO, Publisher<Set<Character>>>() {
          @Override
          public Publisher<Set<Character>> apply(UserPO userPO) {
              char[] chars = userPO.getMobilePhone().toCharArray();
              Set<Character> phones = new HashSet<>();
              for (char aChar : chars) {
                  phones.add(aChar);
              }
              return Flux.just(phones);
          }
      }).subscribe(new Consumer<Set<Character>>() {
          @Override
          public void accept(Set<Character> characters) {
              System.out.println(characters.toString());
          }
      });
      // 输出的结果
      [1, 2, 3]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoYW5nZTk4NzY1NDMyMQ_size_16_color_FFFFFF_t_70 1][]

*  flatmap: 从方法参数来看，将一个事件转换为一个发射器（生产者）；
    
     *  第一个参数：源事件（UserPO）
     *  第二个参数： 转成为发射什么数据类型的发射器（这里发射Set集合）
 *  flatmap 和map使用类型，map返回一个具体对象类型，flatmap更加灵感，返回一个新的被观察者；
 *  他把上游的UserPO对象，转换为了一个发射Set的被被观察者，可以是任意类型，我这里由于需求实现，转换为Set集合；
 *  **需要注意的是flatmap的发射是无序发射的，** concatMap 则是有序的；

### zipwith操作符 ###

需求：不仅要查询用户信息，还要查询用户的收获地址列表；

// 模仿数据库查询
            List<UserPO> userList = new ArrayList<>();
            UserPO userPO = new UserPO();
            userPO.setName("张三");
            userPO.setMobilePhone("1321111111");
            userList.add(userPO);
            // 模仿数据查询地址列表
            EmitterProcessor<List<String>> emitterProcessor = EmitterProcessor.create();
            List<String> addressList = new ArrayList<>();
            addressList.add("北京");
            addressList.add("上海");
            addressList.add("广告");
            emitterProcessor.onNext(addressList);
            // 对查询出来的数据进行逻辑处理
            Flux.fromIterable(userList).map(new Function<UserPO, UserVO>() {
                @Override
                public UserVO apply(UserPO userPO) {
                    UserVO userVO = new UserVO();
                    userVO.setName(userPO.getName());
                    String mobilePhone = userPO.getMobilePhone();
                    StringBuilder sb = new StringBuilder();
                    // 电话号码隐藏处理
                    sb.append(mobilePhone, 0, 3).append("***").append(mobilePhone.substring(mobilePhone.length() - 3));
                    userVO.setPhone(sb.toString());
                    return userVO;
                }
            }).zipWith(emitterProcessor, new BiFunction<UserVO, List<String>, UserVO>() {
                @Override
                public UserVO apply(UserVO userVO, List<String> strings) {
                    userVO.setShippingAddress(strings);
                    return userVO;
                }
            }).subscribe(new Consumer<UserVO>() {
                @Override
                public void accept(UserVO userVO) {
                    System.out.println(userVO);
                }
            });
            // 输出结果
            UserVO(name=张三, phone=132***111, shippingAddress=[北京, 上海, 广告])

*  zipwith 作用就是将两个对象合并和一个对象，在进行传递；
    
     *  第一个参数: 我们传递的源事件
     *  第二个参数：新产生的数据事件
     *  第三个参数：将两个事件源合并在一起，产生一个新的事件源
 *  zipwith在使用上有问题，很鸡肋

### mergeWith 操作符 ###

*  将一个事件插入到原来的事件中；
 *  类似于将List集合中插入一条数据，但是插入在List集合的末尾
 *  和zipwith不同的是，他不与源事件合并，而是插入新的事件；

### collect 操作符 ###

已经上面使用过了，就是做收集；

### reduce 操作符 ###

Flux.just(1,2,3,4,5).reduce(new BiFunction<Integer, Integer, Integer>() {
           @Override
           public Integer apply(Integer integer, Integer integer2) {
               System.out.println(integer+"==过程1");
               System.out.println(integer2+"===过程2");
               return integer+ integer2;
           }
       }).subscribe(new Consumer<Integer>() {
           @Override
           public void accept(Integer integer) {
               System.out.println(integer+"==结果");
           }
       });
       // 输出结果
        1==过程1
    	2===过程2
    	3==过程1
    	3===过程2
    	6==过程1
    	4===过程2
    	10==过程1
    	5===过程2
    	15==结果

*  reduce 将所有数据聚合在一起之后，在传给下游流水线，最后只传递了一个事件给下游；

### buffer 操作符 ###

Flux.just(1,2,3,4,5,6,7).buffer(3).subscribe(new Consumer<List<Integer>>() {
         @Override
           public void accept(List<Integer> integers) {
               System.out.println(integers.toString());
           }
       });
       // 输出结果
       [1, 2, 3]
       [4, 5, 6]
       [7]

*  buffer 就将单个数据分组聚合，3就是几个事件合并为一个组，然后发送出去的是一个List集合；如果最后的数据撮不成3了，就使用余数存放到一个List集合进行发送；

### groupBy ###

Flux.just(1,2,3,4,5,6,7,8).groupBy(new Function<Integer, Integer>() {
         @Override
         public Integer apply(Integer integer) {
             return integer % 3;
         }
     }).subscribe(new Consumer<GroupedFlux<Integer, Integer>>() {
         @Override
         public void accept(GroupedFlux<Integer, Integer> integerIntegerGroupedFlux) {
             integerIntegerGroupedFlux.subscribe(new Consumer<Integer>() {
                 @Override
                 public void accept(Integer integer) {
                     System.out.println("分组id="+integerIntegerGroupedFlux.key()+" ==值=="+integer);
                 }
             });
         }
     });
     // 输出结果
    分组id=1 ==值==1
    分组id=2 ==值==2
    分组id=0 ==值==3
    分组id=1 ==值==4
    分组id=2 ==值==5
    分组id=0 ==值==6
    分组id=1 ==值==7
    分组id=2 ==值==8

*  groupBy 将事件按照某种规则进行分组，同时将分组id传递给下游；分组之后传递给下游的是一个被观察者（生产者）；
 *  与buffer组合发送不同，这里还是一个个事件往下传递；

## 事件处理 ##

### handler ###

Flux.just("15", "26", "").filter(new Predicate<String>() {
          @Override
          public boolean test(String s) {
              if (s == null || "".equals(s)) {
                  return false;
              }
              return true;
          }
      }).handle(new BiConsumer<String, SynchronousSink<Integer>>() {
          @Override
          public void accept(String s, SynchronousSink<Integer> stringSynchronousSink) {
              stringSynchronousSink.next(Integer.valueOf(s));
          }
      }).subscribe(new Consumer<Integer>() {
          @Override
          public void accept(Integer integer) {
              System.out.println("内容="+integer);
          }
      });
      // 输出结果
      内容=15
       内容=26

*  handler ：在事件传递的中间环节做数据的处理，处理过程中如果产生了多个新的事件，通过发射器继续将新事件往下传递；
 *  需要注意的是，handler对源事件进行处理之后传递一个新的事件，每次只能传递一个不能多传；

**handler 和flatMap的区别**

*  handler往下游传递的事件数量不能变
 *  flatmap 接受到源事件之后，可以传递N个新事件，事件数量是变动的；

### doOnNext ###

Flux.just("15", "26", "").filter(new Predicate<String>() {
           @Override
            public boolean test(String s) {
                if (s == null || "".equals(s)) {
                    return false;
                }
                return true;
            }
        }).doOnNext(new Consumer<String>() {
            @Override
            public void accept(String s) {
                System.out.println(s);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String integer) {
                System.out.println("内容="+integer);
            }
        });
        // 输出结果
        内容=11
    	内容=22

*  doOnNext ：也是事件的中间处理流程，它不能改成源事件类型以及发射新的事件；
 *  即使你在中间改变了传递来值，它不会改变源事件中的值；
 *  如果对象引用类型，如传递UserVO对象，doOnNext对UserVO属性name修改，它传递name就会发生变化；

### doOnError ###

*  doOnError 使用和doOnNext相似，就是出现错误后，中间逻辑处理；

### doOnComplete ###

*  doOnComplete 使用和doOnNext相似，整个事件流完成之后应该做些什么事情；

## 重试 ##

### retry 操作符 ###

Flux.just(1,2).handle(new BiConsumer<Integer, SynchronousSink<Integer>>() {
                @Override
                public void accept(Integer integer, SynchronousSink<Integer> stringSynchronousSink) {
                    if(integer == 1){
                        stringSynchronousSink.error(new RuntimeException("错误1"));
                    } else {
                        stringSynchronousSink.next(integer+10);
                    }
                }
            }).retry(2, new Predicate<Throwable>() {
                @Override
                public boolean test(Throwable throwable) {
                    System.out.println("retry重试："+throwable.getMessage());
                    if(throwable instanceof RuntimeException){
                        RuntimeException re = (RuntimeException) throwable;
                        System.out.println("错误内容:"+re.getMessage());
                        return true;
                    }
                    return false;
                }
            }).onErrorResume(new Function<Throwable, Publisher<Integer>>() {
                @Override
                public Publisher<Integer> apply(Throwable throwable) {
                    System.out.println("用0替代");
                    return Flux.just(0);
                }
            }).subscribe(new Consumer<Integer>() {
                @Override
                public void accept(Integer s) {
                    System.out.println("内容:"+s);
                }
            });
            // 输出内容：
            retry重试：错误1
    		错误内容:错误1
    		retry重试：错误1
    		错误内容:错误1
    		用0替代
    		内容:0

*  retry :当我们的处理流程中出现错误时，是否发起重试；
    
     *  第一个参数：最大重试次数
     *  第二个参数：出现错误之后是否重试返回true:重试，false不重试；

### onErrorResume操作符 ###

在捕获到错误之后，我们应该做点什么错误处理的逻辑，返回的是一个发射器；我这里没有做错误的逻辑处理，直接返回Flux.just(0),在项目中可以根据具体需要做处理；

[Spring Reactor]: https://blog.csdn.net/change987654321/article/details/110393508
[Spring Reactor parallel_]: https://blog.csdn.net/change987654321/article/details/110471274
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoYW5nZTk4NzY1NDMyMQ_size_16_color_FFFFFF_t_70]: /images/20221120/3790ad7f1dfc4b1d8d3c0b1d58366382.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoYW5nZTk4NzY1NDMyMQ_size_16_color_FFFFFF_t_70 1]: /images/20221120/5a9b2b70453c48eaaf7c929d02103646.png