Spring Webflux - R2dbc：如何在迭代结果集时运行子查询并更新值

如何解决Spring Webflux - R2dbc：如何在迭代结果集时运行子查询并更新值

我是 Reactive 存储库和 webflux 的新手。我正在从 DB 获取数据列表，使用 map() 对其进行迭代以构建 DTO 类对象，在此过程中我需要运行另一个查询以获取计数值并更新相同的 DTO 对象。当我尝试如下时，计数设置为空

@Repository
public class CandidateGroupCustomRepo {
  public Flux<CandidateGroupListDTO> getList(BigInteger userId){
        final String sql = "SELECT gp.CANDIDATE_GROUP_ID,gp.NAME,gp.GROUP_TYPE   \n" +
                ",gp.CREATED_DATE,cd.DESCRIPTION STATUS,COUNT(con.CANDIDATE_GROUP_ID)\n" +
                "                             FROM  ........" +
                "                             WHERE gp.CREATED_BY_USER_ID = :userId  GROUP BY gp.CANDIDATE_GROUP_ID,cd.DESCRIPTION";
        return dbClient.execute(sql)
                .bind("userId",userId)
                .map(row ->{
                            CandidateGroupListDTO info = new CandidateGroupListDTO();
                            info.setGroupId(row.get(0,BigInteger.class));
                            info.setGroupName(row.get(1,String.class)) ;
                            info.setGroupType(row.get(2,String.class));
                            info.setCreatedDate( row.get(3,LocalDateTime.class));
                            info.setStatus(row.get(4,String.class));

                            if(info.getGroupType().equalsIgnoreCase("static")){
                                info.setContactsCount(row.get(5,BigInteger.class));
                            }else{
                getGroupContactCount(info.getGroupId()).subscribe(count ->{
                    System.out.println(">>>>>"+count);
                    info.setContactsCount(count);
                
                        });
                            }
                            return info;
                            }
                        )
                .all() ;
    }
    
    Mono<BigInteger> getGroupContactCount(BigInteger groupId){
            final String sql = "SELECT 3 WHERE :groupId IS NOT NULL;";
            return dbClient.execute(sql)
                    .bind("groupId",groupId)
                    .map(row -> {
                        System.out.println(row.get(0,BigInteger.class));
                        return row.get(0,BigInteger.class);
                    }  ).one();
    }
    
}

当我调用 getGroupContactCount 时，我试图从 Mono<BigInteger> 中提取计数并将其设置在我的 DTO 中...... .

解决方法

您在中间调用 subscribe，这实际上是阻塞的。订阅者通常是最终消费者，我猜您的 spring 应用程序不是，最终消费者很可能是发起呼叫的网页。您的服务器是生产者。

调用数据库，flatMap 并返回。

return dbClient.execute(sql)
    .bind("userId",userId)
    .flatMap(row ->{
        CandidateGroupListDTO info = new CandidateGroupListDTO();
        info.setGroupId(row.get(0,BigInteger.class));
        info.setGroupName(row.get(1,String.class)) ;
        info.setGroupType(row.get(2,String.class));
        info.setCreatedDate( row.get(3,LocalDateTime.class));
        info.setStatus(row.get(4,String.class));

        if(info.getGroupType().equalsIgnoreCase("static")){
            return Mono.just(info.setContactsCount(row.get(5,BigInteger.class)));
        } else {
            return getGroupContactCount(info.getGroupId()).flatMap(count -> {
                info.setContactsCount(count);
                return Mono.just(info)
            });
        }
    }).all();

如果顺序很重要，请使用 map，否则尝试使用 flatMap 进行异步工作。