public class DynamicDeserializer implements Deserializer<Object> {
    private Map<String, Deserializer<?>> topicDeserializers;

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // 初始化主題與反序列化器的映射關系
        topicDeserializers = new HashMap<>();
        topicDeserializers.put("user-topic", new JsonDeserializer<>(User.class));
        topicDeserializers.put("log-topic", new StringDeserializer());
    }

    @Override
    public Object deserialize(String topic, byte[] data) {
        Deserializer<?> deserializer = topicDeserializers.get(topic);
        if (deserializer == null) {
            throw new IllegalArgumentException("Unsupported topic: " + topic);
        }
        return deserializer.deserialize(topic, data);
    }

    @Override
    public void close() {
        topicDeserializers.values().forEach(Deserializer::close);
    }
}

@Configuration
public class KafkaConfig {

    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "dynamic-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        
        // 關鍵配置：使用自定義動態(tài)反序列化器
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, DynamicDeserializer.class);
        
        // 信任所有包（僅測試環(huán)境使用，生產(chǎn)環(huán)境應限制）
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

@Component
public class KafkaConsumer {

    @KafkaListener(topics = {"user-topic", "log-topic"})
    public void handleMessage(
            @Payload Object payload,
            @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        
        if ("user-topic".equals(topic)) {
            User user = (User) payload;
            System.out.println("Received User: " + user.getName());
        } else if ("log-topic".equals(topic)) {
            String log = (String) payload;
            System.out.println("Received Log: " + log);
        }
    }
}

@Bean
public ConsumerFactory<String, Object> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    // 使用錯誤處理反序列化器包裝
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
    props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, DynamicDeserializer.class);
    return new DefaultKafkaConsumerFactory<>(props);
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory(
        KafkaTemplate<String, Object> kafkaTemplate) {
    
    ConcurrentKafkaListenerContainerFactory<String, Object> factory = 
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    
    // 配置錯誤處理器：重試3次后發(fā)送到死信隊列
    DefaultErrorHandler errorHandler = new DefaultErrorHandler(
        new DeadLetterPublishingRecoverer(kafkaTemplate),
        new FixedBackOff(1000L, 3L)
    );
    factory.setCommonErrorHandler(errorHandler);
    return factory;
}

kafka:
  deserializers:
    user-topic: com.example.UserDeserializer
    log-topic: org.apache.kafka.common.serialization.StringDeserializer

@Value("#{${kafka.deserializers}}")
private Map<String, String> deserializerMappings;

public void configure(Map<String, ?> configs, boolean isKey) {
    topicDeserializers = new HashMap<>();
    deserializerMappings.forEach((topic, deserializerClass) -> {
        try {
            Deserializer<?> deserializer = (Deserializer<?>) Class.forName(deserializerClass).newInstance();
            topicDeserializers.put(topic, deserializer);
        } catch (Exception e) {
            throw new RuntimeException("Failed to initialize deserializer for topic: " + topic, e);
        }
    });
}