如何在Spring Cloud Stream函数式模型中迁移@Valid验证功能？

此前Spring Cloud Stream中可通过@StreamListener结合@Valid注解实现消息参数验证，但自2020年10月6日起@StreamListener被正式弃用，相关示例也已移除。

旧实现方式：

@StreamListener(Processor.INPUT)
@SendTo(Processor.OUTPUT)
public VoteResult handle(@Valid Vote vote) {
  return votingService.record(vote);
}

新的函数式实现方式：

public Function<Vote, VoteResult> handle() {
  return vote -> votingService.record(vote);
}

需迁移上述参数验证功能至新的函数式模型中。

实现步骤 #

1. 引入验证依赖 #

确保项目中已引入Spring Validation依赖（Spring Boot项目可直接引入以下依赖）：

<!-- Maven 依赖示例 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-validation</artifactId>
</dependency>

2. 函数式组件中实现验证逻辑 #

函数式模型无法直接通过@Valid注解触发参数验证，需手动整合Spring Validation能力，以下两种方式任选其一：

方式一：手动注入Validator校验参数

直接在函数逻辑中调用Validator完成参数验证，自定义异常处理逻辑：

import jakarta.validation.Validator;
import jakarta.validation.ConstraintViolation;
import org.springframework.stereotype.Component;
import java.util.Set;
import java.util.function.Function;

@Component
public class VoteFunction {

    private final Validator validator;
    private final VotingService votingService;

    // 构造注入Validator与业务服务
    public VoteFunction(Validator validator, VotingService votingService) {
        this.validator = validator;
        this.votingService = votingService;
    }

    public Function<Vote, VoteResult> handle() {
        return vote -> {
            // 执行参数验证
            Set<ConstraintViolation<Vote>> violations = validator.validate(vote);
            if (!violations.isEmpty()) {
                // 拼接验证错误信息并抛出异常
                String errorMsg = violations.stream()
                        .map(ConstraintViolation::getMessage)
                        .reduce((msg1, msg2) -> msg1 + ", " + msg2)
                        .orElse("参数格式错误");
                throw new IllegalArgumentException(errorMsg);
            }
            // 验证通过后执行业务逻辑
            return votingService.record(vote);
        };
    }
}

方式二：通过@Validated让业务服务自动校验

将验证逻辑封装到业务服务层，利用@Validated注解触发Spring AOP自动验证：

import jakarta.validation.Valid;
import org.springframework.validation.annotation.Validated;
import org.springframework.stereotype.Service;

@Service
@Validated
public class VotingService {

    public VoteResult record(@Valid Vote vote) {
        // 执行业务逻辑
    }
}

此时函数式方法可直接调用业务方法，验证失败时会自动抛出ConstraintViolationException：

import org.springframework.stereotype.Component;
import java.util.function.Function;

@Component
public class VoteFunction {

    private final VotingService votingService;

    public VoteFunction(VotingService votingService) {
        this.votingService = votingService;
    }

    public Function<Vote, VoteResult> handle() {
        return votingService::record;
    }
}

3. 全局处理验证异常（可选） #

可通过Spring Cloud Stream的错误处理机制统一捕获验证异常，做日志记录或死信消息处理：

import jakarta.validation.ConstraintViolationException;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.support.ErrorMessage;
import java.util.function.Consumer;

@Configuration
public class ErrorHandlingConfig {

    @Bean
    public Consumer<ErrorMessage> validationErrorHandler() {
        return errorMessage -> {
            Throwable cause = errorMessage.getPayload();
            if (cause instanceof ConstraintViolationException violationEx) {
                // 记录验证失败日志
                System.err.println("消息参数验证失败: " + violationEx.getMessage());
                // 可扩展死信消息转发逻辑
            }
        };
    }
}

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