spring-data-redis泛型ObjectRecord反序列化及多流监听配置问题

当前反序列化仅返回类全限定名、属性丢失的核心原因是：
消费者配置的监听目标类型为ObjectRecord<String, Object>，泛型类型擦除后，Jackson2HashMapper在反序列化时仅能识别Hash结构中存储的@class类型标识字段，无法获知目标对象的实际属性结构，更无法解析MyMessage上的泛型参数类型，因此不会完成其他字段的映射。

推荐使用手动反序列化的方案，兼容性最好，也能完美支持泛型处理，修改步骤如下：

1. 调整监听器容器的泛型配置，不再让容器自动做Object类型转换，直接接收原始Map结构的消息：

@Bean
StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, Object, Object>> hashContainerOptions() {
    return StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
            .pollTimeout(Duration.ofSeconds(1))
            .build();
}

2. 同步修改容器的泛型声明，在消息监听逻辑中拿到原始Map后，手动构造带泛型信息的目标类型完成反序列化：

@Bean
StreamMessageListenerContainer<String, MapRecord<String, Object, Object>> hashContainer(
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, Object, Object>> options,
        RedisConnectionFactory redisConnectionFactory,
        RedisTemplate<String, Object> redisTemplate) {

    var container = StreamMessageListenerContainer.create(redisConnectionFactory, options);
    container.start();
    String group = "default-group";
    String key = "my-stream";
    try {
        redisTemplate.opsForStream().createGroup(key, group);
    } catch (Exception ignore) {}

    // 初始化和生产者配置对齐的ObjectMapper
    ObjectMapper objectMapper = new ObjectMapper();
    objectMapper.activateDefaultTyping(LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL);

    container.receiveAutoAck(
            Consumer.from(group, "default-consumer"),
            StreamOffset.create(key, ReadOffset.lastConsumed()),
            message -> {
                Map<Object, Object> valueMap = message.getValue();
                // 明确构造泛型类型 MyMessage<User>
                JavaType targetType = objectMapper.getTypeFactory()
                        .constructParametricType(MyMessage.class, User.class);
                MyMessage<User> msg = objectMapper.convertValue(valueMap, targetType);
                log.info("receive message stream:{}, id:{} value:{}", message.getStream(), message.getId(), msg);
            }
    );
    return container;
}

生产者侧原有写入逻辑不需要调整，写入Redis后的存储结构如下：

如果不想手动做反序列化，也可以直接发送ObjectRecord类型的消息，去掉Jackson2HashMapper转Map的步骤，直接传入实体对象，让RedisTemplate配置的JSON序列化器统一处理类型信息，但这种方式对泛型的支持不如手动反序列化灵活。

不需要为不同消息类型的Stream创建多个独立的StreamMessageListenerContainer实例。
单个容器实例支持同时绑定多个Stream的监听任务，你只需要在同一个容器上多次调用receiveAutoAck/receive方法，为每个Stream配置独立的消费组、消费偏移量、以及对应类型的消息反序列化和处理逻辑即可。容器会统一管理调度线程、拉取任务，资源占用比多容器方案更低。

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