KafkaTemplate 泛型选择：多类型消息发送的最佳实践

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.kafka.support.serializer.JsonSerializer; // 假设使用JSON序列化

@Configuration
public class KafkaProducerConfig {

    // 生产者的通用配置
    public Map<String, Object> producerConfiguration() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // 值序列化器配置为JsonSerializer，以便处理不同类型的Java对象
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        // 可选：如果需要对JSON序列化器进行特定配置
        // props.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false); 
        return props;
    }

    // 定义一个通用的 ProducerFactory
    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfiguration());
    }

    // 定义一个通用的 KafkaTemplate
    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.kafka.support.serializer.JsonSerializer;

// 假设有 Address 和 Person 两个消息类
class Address { /* ... */ }
class Person { /* ... */ }

@Configuration
public class SpecificKafkaTemplatesConfig {

    public Map<String, Object> producerConfiguration() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // 假设所有值都用 JsonSerializer 序列化
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        return props;
    }

    // 为 Address 类型定义 ProducerFactory 和 KafkaTemplate
    @Bean
    public ProducerFactory<String, Address> addressProducerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfiguration());
    }

    @Bean
    public KafkaTemplate<String, Address> addressKafkaTemplate() {
        return new KafkaTemplate<>(addressProducerFactory());
    }

    // 为 Person 类型定义 ProducerFactory 和 KafkaTemplate
    @Bean
    public ProducerFactory<String, Person> personProducerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfiguration());
    }

    @Bean
    public KafkaTemplate<String, Person> personKafkaTemplate() {
        return new KafkaTemplate<>(personProducerFactory());
    }

    // 也可以直接在 KafkaTemplate 构造函数中创建 ProducerFactory，减少 Bean 定义
    @Bean
    public KafkaTemplate<String, AnotherType> anotherTypeKafkaTemplate() {
        return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(producerConfiguration()));
    }
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

// 假设有 Address 和 Person 两个消息类
class Address {
    private String street;
    private String city;
    // Getters and Setters
}
class Person {
    private String name;
    private int age;
    // Getters and Setters
}

@Service
public class MessageSenderService {

    // 注入一个泛型为 <String, Object> 的 KafkaTemplate
    // 实际上，Spring Boot 自动配置的 KafkaTemplate<K, V> 允许你以任何具体类型注入
    @Autowired
    private KafkaTemplate<String, Object> genericKafkaTemplate;

    // 也可以直接注入特定类型的 KafkaTemplate，Spring Boot 会自动匹配
    @Autowired
    private KafkaTemplate<String, Address> addressKafkaTemplate;

    @Autowired
    private KafkaTemplate<String, Person> personKafkaTemplate;

    public void sendAddress(String topic, String key, Address address) {
        // 使用泛型 KafkaTemplate 发送
        genericKafkaTemplate.send(topic, key, address);
        System.out.println("Sent address via generic template: " + address);

        // 或者使用特定类型注入的 KafkaTemplate 发送
        addressKafkaTemplate.send(topic, key, address);
        System.out.println("Sent address via specific template: " + address);
    }

    public void sendPerson(String topic, String key, Person person) {
        // 使用泛型 KafkaTemplate 发送
        genericKafkaTemplate.send(topic, key, person);
        System.out.println("Sent person via generic template: " + person);

        // 或者使用特定类型注入的 KafkaTemplate 发送
        personKafkaTemplate.send(topic, key, person);
        System.out.println("Sent person via specific template: " + person);
    }
}