SpringBoot实现Kafka动态反序列化的完整代码

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);
        
        // 关键配置：使用自定义动态反序列化器
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, DynamicDeserializer.class);
        
        // 信任所有包（仅测试环境使用，生产环境应限制）
        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次后发送到死信队列
    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);
        }
    });
}