springkafka自定义 Kafka 的生产者和消费者工厂

Spring Kafka 提供了方便的方式来创建和管理自定义的 Kafka 生产者和消费者。它允许你通过实现 `KafkaProducerFactory` 和 `KafkaConsumerFactory` 接口来自定义生产和消费的行为。

对于自定义生产者工厂 (`KafkaProducerFactory`):

@Configuration
public class CustomProducerConfig {

    @Bean
    public KafkaProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        // 自定义配置，例如设置acks、retries等
        configProps.put(ProducerConfig.ACKS_CONFIG, "all");
        configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, YourCustomKeySerializer.class);
        configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, YourCustomValueSerializer.class);

        return new DefaultKafkaProducerFactory<>(configProps, createProducerConfig());
    }

    private ProducerConfig createProducerConfig() {
        return new ProducerConfig();
    }
}

这里的 `YourCustomKeySerializer` 和 `YourCustomValueSerializer` 是你自定义的序列化器实现。

对于自定义消费者工厂 (`KafkaConsumerFactory`)：

@Configuration
public class CustomConsumerConfig {

    @Bean
    public ConcurrentKafkaConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        // 自定义配置，例如设置bootstrap servers、group id等
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, "your-consumer-group");

        return new ConcurrentKafkaConsumerFactory<>(createConsumerConfig(), keyDeserializerClass(), valueDeserializerClass());
    }

    private ConsumerConfig createConsumerConfig() {
        return new ConsumerConfig();
    }
    
    // 这里分别定义自定义的key和value反序列化器
    private Class<? extends Deserializer<String>> keyDeserializerClass() { return YourCustomKeyDeserializer.class; }
    private Class<? extends Deserializer<String>> valueDeserializerClass() { return YourCustomValueDeserializer.class; }
}

这样，Spring Kafka会根据你的工厂配置来实例化对应的生产者和消费者。记得提供完整的类路径，并确保它们实现了Kafka的序列化和反序列化接口。