Reactor：用Mono用户名过滤Flux课程无匹配结果排查

我使用Cassandra作为后端，表结构如下：

CREATE TABLE main.courses_by_user (
    course_creator text PRIMARY KEY,
    courseid timeuuid,
    description text,
    enrollmentkey text
) WITH additional_write_policy = '99PERCENTILE'
    AND bloom_filter_fp_chance = 0.01
    AND caching = {'keys': 'ALL', 'rows_per_partition': 'NONE'}
    AND comment = ''
    AND compaction = {'class': 'org.apache.cassandra.db.compaction.UnifiedCompactionStrategy'}
    AND compression = {'chunk_length_in_kb': '64', 'class': 'org.apache.cassandra.io.compress.LZ4Compressor'}
    AND crc_check_chance = 1.0
    AND default_time_to_live = 0
    AND gc_grace_seconds = 864000
    AND max_index_interval = 2048
    AND memtable_flush_period_in_ms = 0
    AND min_index_interval = 128
    AND read_repair = 'BLOCKING'
    AND speculative_retry = '99PERCENTILE';

表中数据：

course_creator | courseid                             | description | enrollmentkey
----------------+--------------------------------------+-------------+---------------
        rascall | b7757e80-0c24-11ed-aec5-23fe9d87e512 |   Cosmology |        hubble
       dibiasky | b7757e81-0c24-11ed-aec5-23fe9d87e512 |   astronomy |    thebigbang
  michaeljburry | b7753060-0c24-11ed-aec5-23fe9d87e512 |         Lol |         enter
   Noam Chomsky | 6c1a4800-09ac-11ed-ada9-83d934863d60 |          Hi |           Bye

我尝试用包含用户名的Mono过滤存储多门课程的Flux，使用zipWith方法将二者配对，代码如下：

public Flux<CourseByCreator> getMyCourses(Principal placeholder){
        Mono<User> principal = Mono.just(new User("dibiasky", "whatifitsnotequal","Kate" ,"Dibiasky"));
        return allCreatedCourses = this.courseByCreatorRepository.findAll()
                .zipWith(principal)
                 .flatMap(tuple -> {
                     if(tuple.getT1().getCourseCreator().equals(tuple.getT2().getUsername())){
                         System.out.println(tuple.getT2().getUsername());
                         return Flux.just(tuple.getT1());
                     }else{
                         return Flux.empty();
                     }
                 }).log();
}

明明存在与用户名匹配的课程创建者，但返回结果为空，请问我哪里操作错误？

错误原因 #

你误用了zipWith操作符：

• zipWith的逻辑是一对一配对，它会把Flux的第一个元素和Mono的唯一元素配对，后续Flux的元素找不到对应的Mono元素（因为Mono仅包含一个值），不会被处理。
• 你的课程Flux有4个元素，zipWith只会处理第一个（rascall的课程），和dibiasky的用户对象配对后因不匹配返回空，后续匹配的课程根本没机会被处理，最终结果为空。

正确写法 #

应该先获取Mono<User>的值，再用该值过滤整个Flux<CourseByCreator>，以下是两种可行方案：

方案一：使用flatMapMany

public Flux<CourseByCreator> getMyCourses(Principal placeholder){
    Mono<User> principalMono = Mono.just(new User("dibiasky", "whatifitsnotequal","Kate" ,"Dibiasky"));
    
    // 先拿到用户对象，再过滤课程列表
    return principalMono.flatMapMany(user -> 
        this.courseByCreatorRepository.findAll()
            .filter(course -> course.getCourseCreator().equals(user.getUsername()))
    ).log();
}

方案二：用withLatestFrom关联用户数据

public Flux<CourseByCreator> getMyCourses(Principal placeholder){
    Mono<String> usernameMono = Mono.just(new User("dibiasky", "whatifitsnotequal","Kate" ,"Dibiasky"))
                                    .map(User::getUsername);
    
    return this.courseByCreatorRepository.findAll()
            .withLatestFrom(usernameMono, (course, username) -> Tuple2.of(course, username))
            .filter(tuple -> tuple.getT1().getCourseCreator().equals(tuple.getT2()))
            .map(Tuple2::getT1)
            .log();
}

额外性能优化 #

从Cassandra表结构来看，course_creator是主键，完全可以直接通过主键查询该用户的课程，而非全表查询后过滤，性能会大幅提升。在Repository中添加方法：

Flux<CourseByCreator> findByCourseCreator(String courseCreator);

然后直接调用：

public Flux<CourseByCreator> getMyCourses(Principal placeholder){
    Mono<User> principalMono = Mono.just(new User("dibiasky", "whatifitsnotequal","Kate" ,"Dibiasky"));
    
    return principalMono.flatMapMany(user -> 
        this.courseByCreatorRepository.findByCourseCreator(user.getUsername())
    ).log();
}

内容的提问来源于stack exchange，提问作者Kantenburg