如何使用@SqsListener处理多对象类型？求SQS版@RabbitHandler等效方案

Spring Cloud AWS的@SqsListener本身没有类似RabbitMQ @RabbitHandler的自动类型匹配能力——当多个方法监听同一个队列时，消息会被随机分配给任意方法，和消息内容无关。要实现按对象类型匹配处理方法，需要手动实现消息路由逻辑，以下是两种可行方案：

方案一：基于消息头部的类型路由 #

通过在发送消息时添加类型标识头部，接收端解析头部后转成对应实体类，再分发到专属处理器。

1. 定义消息处理器 #

将不同类型的消息处理逻辑拆分到独立的处理器类中：

@Component
public class TestModelHandler {
    private static final Logger LOGGER = LoggerFactory.getLogger(TestModelHandler.class);

    public void handle(TestModel model) {
        LOGGER.info("Received a TestModel.");
    }
}

@Component
public class AnotherTestModelHandler {
    private static final Logger LOGGER = LoggerFactory.getLogger(AnotherTestModelHandler.class);

    public void handle(AnotherTestModel model) {
        LOGGER.info("Received AnotherTestModel.");
    }
}

2. 实现统一路由监听 #

创建一个唯一的@SqsListener方法接收所有消息，解析类型头部后路由到对应处理器：

@Service
public class SqsMessageRouter {
    private static final Logger LOGGER = LoggerFactory.getLogger(SqsMessageRouter.class);
    private final ObjectMapper objectMapper;
    private final Map<Class<?>, Consumer<?>> handlerMap;

    // 构造注入处理器，建立类型与处理器的映射
    public SqsMessageRouter(ObjectMapper objectMapper, TestModelHandler testHandler, AnotherTestModelHandler anotherHandler) {
        this.objectMapper = objectMapper;
        this.handlerMap = Map.of(
                TestModel.class, testHandler::handle,
                AnotherTestModel.class, anotherHandler::handle
        );
    }

    @SqsListener("some-queue")
    public void routeMessage(String messageBody, @Header("type") String type) throws IOException {
        // 根据头部类型匹配目标类
        Class<?> targetClass = switch (type) {
            case "test-model" -> TestModel.class;
            case "another-test-model" -> AnotherTestModel.class;
            default -> throw new IllegalArgumentException("Unknown message type: " + type);
        };

        // 反序列化消息体
        Object payload = objectMapper.readValue(messageBody, targetClass);
        // 调用对应处理器
        Consumer<?> handler = handlerMap.get(targetClass);
        if (handler != null) {
            ((Consumer<Object>) handler).accept(payload);
        } else {
            LOGGER.warn("No handler found for message type: {}", type);
        }
    }
}

3. 发送消息时添加类型头部 #

发送消息时通过SqsTemplate设置type头部：

sqsTemplate.send("some-queue", Message.builder()
        .body(objectMapper.writeValueAsString(testModel))
        .header("type", "test-model")
        .build());

方案二：基于Jackson多态反序列化的路由 #

利用Jackson的多态解析功能，在消息体中嵌入类型标识，自动反序列化为对应子类后分发。

1. 定义消息父类/接口 #

创建所有消息类型的父接口，并配置Jackson多态规则：

@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "@type")
@JsonSubTypes({
        @JsonSubTypes.Type(value = TestModel.class, name = "test-model"),
        @JsonSubTypes.Type(value = AnotherTestModel.class, name = "another-test-model")
})
public interface BaseMessage {
}

// 实现父接口
public class TestModel implements BaseMessage {
    // 消息字段
}

public class AnotherTestModel implements BaseMessage {
    // 消息字段
}

2. 实现分发监听 #

创建唯一的@SqsListener方法接收BaseMessage类型，再根据实际类型调用处理器：

@Service
public class SqsMessageDispatcher {
    private static final Logger LOGGER = LoggerFactory.getLogger(SqsMessageDispatcher.class);
    private final TestModelHandler testHandler;
    private final AnotherTestModelHandler anotherHandler;

    public SqsMessageDispatcher(TestModelHandler testHandler, AnotherTestModelHandler anotherHandler) {
        this.testHandler = testHandler;
        this.anotherHandler = anotherHandler;
    }

    @SqsListener("some-queue")
    public void dispatch(BaseMessage message) {
        if (message instanceof TestModel testModel) {
            testHandler.handle(testModel);
        } else if (message instanceof AnotherTestModel anotherTestModel) {
            anotherHandler.handle(anotherTestModel);
        } else {
            LOGGER.warn("Unsupported message type: {}", message.getClass().getName());
        }
    }
}

3. 配置消息转换器 #

确保Spring Cloud AWS使用配置好的Jackson转换器进行反序列化：

@Configuration
public class SqsConfig {
    @Bean
    public MappingJackson2MessageConverter mappingJackson2MessageConverter(ObjectMapper objectMapper) {
        MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
        converter.setObjectMapper(objectMapper);
        converter.setSerializedPayloadClass(String.class);
        converter.setStrictContentTypeMatch(false);
        return converter;
    }

    @Bean
    public QueueMessageHandlerFactory queueMessageHandlerFactory(MappingJackson2MessageConverter converter) {
        QueueMessageHandlerFactory factory = new QueueMessageHandlerFactory();
        factory.setMessageConverters(List.of(converter));
        return factory;
    }
}

发送消息注意事项 #

发送消息时只需正常序列化子类对象，Jackson会自动在JSON中添加@type字段：

sqsTemplate.send("some-queue", objectMapper.writeValueAsString(testModel));

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