Spring Cloud Function在AWS Lambda中Reactive类型转Message报错

问题描述 #

在AWS Lambda中部署Spring Cloud Function的Reactive类型（Flux/Mono）示例时，触发ClassCastException，错误提示reactor.core.publisher.FluxMapFuseable cannot be cast to org.springframework.messaging.Message。本地通过gradlew bootRun运行无异常，不使用Reactive类型时Lambda也能正常工作。疑问：是否需要自定义消息转换器或存在依赖缺失？若需要，能否提供convertFromInternal方法的实现逻辑？

测试代码 #

@SpringBootApplication
public class Application {

  @Bean
  public Function<Flux<String>, Flux<String>> uppercase() {
    return flux -> flux.map(value -> value.toUpperCase());
  }

  public static void main(String[] args) {
    SpringApplication.run(Application.class, args);
  }
}

错误栈信息 #

{
  "errorType": "ClassCastException",
  "errorMessage": "reactor.core.publisher.FluxMapFuseable cannot be cast to org.springframework.messaging.Message",
  "stackTrace": "java.lang.ClassCastException: reactor.core.publisher.FluxMapFuseable cannot be cast to org.springframework.messaging.Message\n\tat org.springframework.cloud.function.adapter.aws.CustomRuntimeEventLoop.eventLoop(CustomRuntimeEventLoop.java:161)\n\tat org.springframework.cloud.function.adapter.aws.CustomRuntimeEventLoop.lambda$run$0(CustomRuntimeEventLoop.java:90)\n\tat java.base@17.0.7/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)\n\tat java.base@17.0.7/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)\n\tat java.base@17.0.7/java.lang.Thread.run(Thread.java:833)\n\tat org.graalvm.nativeimage.builder/com.oracle.svm.core.thread.PlatformThreads.threadStartRoutine(PlatformThreads.java:775)\n\tat org.graalvm.nativeimage.builder/com.oracle.svm.core.posix.thread.PosixPlatformThreads.pthreadStartRoutine(PosixPlatformThreads.java:203)\n"
}

问题原因与解决方案 #

原因分析 #

AWS Lambda的Spring Cloud Function适配器默认期望处理org.springframework.messaging.Message类型的输入输出，但直接返回Reactive类型（Flux/Mono）时，适配器无法自动完成类型转换，导致抛出类型转换异常。本地运行正常是因为Spring Boot Web环境会自动处理Reactive类型的消息适配，而Lambda的自定义运行时适配器未实现该逻辑。

解决方法 #

1. 检查依赖完整性

确保build.gradle（或pom.xml）中包含Spring Cloud Function的Reactive组件及AWS适配器依赖：
Gradle示例：

implementation 'org.springframework.cloud:spring-cloud-function-adapter-aws'
implementation 'org.springframework.cloud:spring-cloud-function-reactive'

2. 自定义消息转换器

实现MessageConverter来处理Reactive类型与Lambda消息格式的转换，核心逻辑如下：

import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.AbstractMessageConverter;
import org.springframework.util.MimeTypeUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class ReactiveMessageConverter extends AbstractMessageConverter {

    public ReactiveMessageConverter() {
        super(MimeTypeUtils.APPLICATION_JSON);
    }

    @Override
    protected boolean supports(Class<?> clazz) {
        return Flux.class.isAssignableFrom(clazz) || Mono.class.isAssignableFrom(clazz);
    }

    @Override
    protected Object convertFromInternal(Message<?> message, Class<?> targetClass, Object conversionHint) {
        // 将Lambda输入的Message负载转换为Reactive类型
        Object payload = message.getPayload();
        if (Flux.class.isAssignableFrom(targetClass)) {
            if (payload instanceof String) {
                return Flux.just((String) payload);
            } else if (payload instanceof String[]) {
                return Flux.fromArray((String[]) payload);
            }
            return Flux.just(payload);
        } else if (Mono.class.isAssignableFrom(targetClass)) {
            return Mono.just(payload);
        }
        return payload;
    }

    @Override
    protected Object convertToInternal(Object payload, MessageHeaders headers, Object conversionHint) {
        // 将Reactive输出转换为Lambda可识别的Message格式
        if (payload instanceof Flux) {
            return ((Flux<?>) payload).collectList().map(list -> 
                org.springframework.messaging.support.MessageBuilder.createMessage(list, headers)
            ).block();
        } else if (payload instanceof Mono) {
            return ((Mono<?>) payload).map(data -> 
                org.springframework.messaging.support.MessageBuilder.createMessage(data, headers)
            ).block();
        }
        return org.springframework.messaging.support.MessageBuilder.createMessage(payload, headers);
    }
}

3. 注册自定义转换器

在Spring配置类中注册该转换器，让AWS适配器使用：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.converter.MessageConverter;

@Configuration
public class LambdaConfig {

    @Bean
    public MessageConverter reactiveMessageConverter() {
        return new ReactiveMessageConverter();
    }
}

4. 调整Function输入输出（可选）

直接将Function的输入输出改为Message包裹的Reactive类型，适配适配器的默认处理逻辑：

@Bean
public Function<Message<Flux<String>>, Message<Flux<String>>> uppercase() {
    return message -> {
        Flux<String> result = message.getPayload().map(String::toUpperCase);
        return org.springframework.messaging.support.MessageBuilder.withPayload(result)
                .copyHeaders(message.getHeaders())
                .build();
    };
}

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