如何在SpringWebFlux中使用zip操作符处理多空值场景

开发者阿泽

2026-7-2

问题分析与解决方案

核心问题

错误使用block()：在响应式流中调用block()会阻塞当前线程，违背Reactor的非阻塞设计，同时破坏整个链式调用的异步流程。
未完整合并所有Mono：当前仅合并了userMono和categoryMono，遗漏了tagListMono、imageName、blogMono三个需要合并的结果。
可空Mono处理不当：部分Mono（如categoryMono、tagListMono）可能为空，直接参与zip会导致流终止，需要先处理为非空流。

修正方案

步骤1：处理可空Mono

对每个可能为空的Mono，使用defaultIfEmpty(null)确保流不会终止，同时保留空值语义：

// 处理Category，为空时返回null
Mono<Category> categoryMono = Mono.justOrEmpty(createBlogRequest.getCategoryId())
        .flatMap(categoryId -> categoryRepository.findById(categoryId)
                .switchIfEmpty(Mono.error(new ResourceNotFoundException("Category not found!"))))
        .defaultIfEmpty(null);

// 处理Tag列表，为空时返回null
Mono<List<Tag>> tagListMono = Mono.justOrEmpty(createBlogRequest.getTagsId())
        .filter(tagsId -> !tagsId.isEmpty())
        .flatMap(tagsId -> tagRepository.findAllById(tagsId)
                .collectList()
                .flatMap(tags -> {
                    if (tags.size() < tagsId.size()) // 原逻辑错误，修正为：查询到的标签少于传入ID则抛出异常
                        return Mono.error(new ResourceNotFoundException("Invalid tag input!"));
                    return Mono.just(tags);
                }))
        .defaultIfEmpty(null);

// 处理图片名称，为空时返回null
Mono<String> imageNameMono = Mono.justOrEmpty(createBlogRequest.getImageFile())
        .flatMap(imageFile -> {
            String name = imageFile.filename() + UUID.randomUUID();
            return Mono.fromRunnable(() -> {
                try {
                    imageFile.transferTo(Paths.get(uploadDir + "/static/" + name));
                } catch (IOException e) {
                    throw new BlogException("Failed to upload image!");
                }
            }).then(Mono.just(name));
        })
        .defaultIfEmpty(null);

// 处理Blog，为空时返回null（新增场景）
Mono<Blog> blogMono = Mono.justOrEmpty(createBlogRequest.getBlogId())
        .flatMap(blogId -> blogRepository.findById(blogId)
                .switchIfEmpty(Mono.error(new ResourceNotFoundException("Blog with id " + blogId + " doesn't exist!"))))
        .defaultIfEmpty(null);

步骤2：合并所有Mono并构建Blog

使用Mono.zip的重载方法合并所有5个Mono，在map中提取元组值构建最终Blog对象，全程保持响应式链式调用，禁止调用block()：

return Mono.zip(userMono, categoryMono, tagListMono, imageNameMono, blogMono)
        .map(tuple -> {
            User user = tuple.getT1();
            Category category = tuple.getT2();
            List<Tag> tags = tuple.getT3();
            String imageName = tuple.getT4();
            Blog existingBlog = tuple.getT5();

            // 构建Blog对象（复用已有Blog或新建）
            Blog blog = existingBlog != null ? existingBlog : new Blog();
            blog.setUser(user);
            blog.setCategory(category);
            blog.setTags(tags);
            blog.setImageName(imageName);
            // 补充其他字段赋值，比如从createBlogRequest获取标题、内容等
            return blog;
        })
        .flatMap(blog -> {
            // 保存或更新Blog
            return blog.getId() != null ? blogRepository.save(blog) : blogRepository.insert(blog);
        })
        .flatMap(savedBlog -> ServerResponse.ok().bodyValue(savedBlog))
        .onErrorResume(e -> {
            // 统一异常处理
            if (e instanceof BlogException || e instanceof ResourceNotFoundException || e instanceof BadCredentialsException) {
                return ServerResponse.badRequest().bodyValue(e.getMessage());
            }
            return ServerResponse.internalServerError().bodyValue("Internal server error!");
        });

完整修正后的代码片段

public Mono<ServerResponse> addOrUpdateBlog(ServerRequest serverRequest) {
    String jwt = serverRequest.headers().firstHeader(HttpHeaders.AUTHORIZATION);
    return serverRequest.body(BodyExtractors.toMultipartData())
            .flatMap(multiValueMap -> {
                if (multiValueMap.containsKey("blogRequest")) {
                    return multiValueMap.getFirst("blogRequest").content().map(dataBuffer -> {
                                byte[] bytes = new byte[dataBuffer.readableByteCount()];
                                dataBuffer.read(bytes);
                                DataBufferUtils.release(dataBuffer);
                                return new String(bytes, StandardCharsets.UTF_8);
                            }).collectList()
                            .flatMap(stringList -> {
                                try {
                                    CreateBlogRequest createBlogRequest = new ObjectMapper().readValue(String.join("", stringList), CreateBlogRequest.class);
                                    if (multiValueMap.containsKey("imageFile"))
                                        createBlogRequest.setImageFile((FilePart) multiValueMap.getFirst("imageFile"));
                                    return Mono.just(createBlogRequest);
                                } catch (JsonProcessingException e) {
                                    return Mono.error(new BlogException("Fail to parse json request!"));
                                }
                            });
                } else {
                    CreateBlogRequest createBlogRequest = new CreateBlogRequest();
                    createBlogRequest.setImageFile((FilePart) multiValueMap.getFirst("imageFile"));
                    return Mono.just(createBlogRequest);
                }
            })
            .flatMap(createBlogRequest -> {
                validationUtils.validateRequest(createBlogRequest);

                Mono<User> userMono = userRepository.findById(tokenService.getUserIdFromToken(jwt))
                        .switchIfEmpty(Mono.error(new BadCredentialsException("Invalid token received!")));

                Mono<Category> categoryMono = Mono.justOrEmpty(createBlogRequest.getCategoryId())
                        .flatMap(categoryId -> categoryRepository.findById(categoryId)
                                .switchIfEmpty(Mono.error(new ResourceNotFoundException("Category not found!"))))
                        .defaultIfEmpty(null);

                Mono<List<Tag>> tagListMono = Mono.justOrEmpty(createBlogRequest.getTagsId())
                        .filter(tagsId -> !tagsId.isEmpty())
                        .flatMap(tagsId -> tagRepository.findAllById(tagsId)
                                .collectList()
                                .flatMap(tags -> {
                                    if (tags.size() < tagsId.size())
                                        return Mono.error(new ResourceNotFoundException("Invalid tag input!"));
                                    return Mono.just(tags);
                                }))
                        .defaultIfEmpty(null);

                Mono<String> imageNameMono = Mono.justOrEmpty(createBlogRequest.getImageFile())
                        .flatMap(imageFile -> {
                            String name = imageFile.filename() + UUID.randomUUID();
                            return Mono.fromRunnable(() -> {
                                try {
                                    imageFile.transferTo(Paths.get(uploadDir + "/static/" + name));
                                } catch (IOException e) {
                                    throw new BlogException("Failed to upload image!");
                                }
                            }).then(Mono.just(name));
                        })
                        .defaultIfEmpty(null);

                Mono<Blog> blogMono = Mono.justOrEmpty(createBlogRequest.getBlogId())
                        .flatMap(blogId -> blogRepository.findById(blogId)
                                .switchIfEmpty(Mono.error(new ResourceNotFoundException("Blog with id " + blogId + " doesn't exist!"))))
                        .defaultIfEmpty(null);

                return Mono.zip(userMono, categoryMono, tagListMono, imageNameMono, blogMono)
                        .map(tuple -> {
                            User user = tuple.getT1();
                            Category category = tuple.getT2();
                            List<Tag> tags = tuple.getT3();
                            String imageName = tuple.getT4();
                            Blog existingBlog = tuple.getT5();

                            Blog blog = existingBlog != null ? existingBlog : new Blog();
                            blog.setUser(user);
                            blog.setCategory(category);
                            blog.setTags(tags);
                            blog.setImageName(imageName);
                            // 补充其他字段赋值逻辑
                            return blog;
                        })
                        .flatMap(blog -> {
                            return blog.getId() != null ? blogRepository.save(blog) : blogRepository.insert(blog);
                        })
                        .flatMap(savedBlog -> ServerResponse.ok().bodyValue(savedBlog))
                        .onErrorResume(e -> {
                            if (e instanceof BlogException || e instanceof ResourceNotFoundException || e instanceof BadCredentialsException) {
                                return ServerResponse.badRequest().bodyValue(e.getMessage());
                            }
                            return ServerResponse.internalServerError().bodyValue("Internal server error!");
                        });
            });
}

关键注意点

禁止在响应式链中调用block()：这会阻塞线程，导致性能下降甚至死锁，应始终用flatMap、map等操作符保持链式调用。
处理可空Mono：使用defaultIfEmpty(null)确保所有参与zip的Mono都能正常发出信号，避免流终止。
异常统一处理：通过onErrorResume集中处理所有异常，返回对应HTTP响应。

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