关于KafkaTemplate泛型参数选型合理性及使用规范的咨询

问题解答 #

1. KafkaTemplate<Object, Object> 是否可用？

完全可行。该泛型定义的兼容性最强，支持发送任意类型的key和value，你当前的代码可以正常运行，唯一的缺点是缺少编译期的类型校验，无法在编码阶段提前发现消息类型不匹配的问题。

2. KafkaTemplate<Object, PlanDTO> 是否是更适配的选型？

是的，更适配你当前的场景。你这个PlanProducer类只负责发送PlanDTO类型的消息，把value的泛型指定为PlanDTO之后，编译期就会自动校验你发送的消息payload类型是否合法，避免误传其他类型的消息导致运行时报错，同时代码里的SendResult、KafkaSendCallback等相关类的泛型也可以同步调整，类型更清晰，也不需要额外的类型转换。

调整后的代码示例：

@Component
public class PlanProducer 
{
    @Autowired
    private KafkaTemplate<Object, PlanDTO> kafkaTemplate;

    public void postPlanToKafka(PlanDTO plan)
    {
            Message<PlanDTO> message = MessageBuilder.withPayload(plan)
            .setHeader(KafkaHeaders.TOPIC, "plan")
            .setHeader(KafkaHeaders.TIMESTAMP, Instant.now().getEpochSecond())
            .build();
            ListenableFuture<SendResult<Object, PlanDTO>> future = kafkaTemplate.send(message);
            future.addCallback(new KafkaSendCallback<Object, PlanDTO>() 
            {
                @Override
                public void onSuccess(SendResult<Object, PlanDTO> result) 
                {
                    RecordMetadata metadata = result.getRecordMetadata();
                    log.info("Message sent successfully with the following information : topic - {} , "
                            + "partition  - {} , offset - {}",metadata.topic(),
                            metadata.partition(),metadata.offset());
                }

                @Override
                public void onFailure(KafkaProducerException e) 
                {
                    log.error("Error posting data to kafka ={}",e);
                }
            });
    }
}

3. KafkaTemplate类型参数定义规则

KafkaTemplate的泛型格式为KafkaTemplate<K, V>，其中K是消息key的类型，V是消息value的类型，定义规则如下：

• 不需要设置消息key的场景下，K可以直接定义为Object，也可以根据实际业务中可能用到的key类型定义为String、Long等具体类型
• V的类型优先匹配当前KafkaTemplate实例要发送的消息payload类型：如果单个实例只发送一种类型的消息，就定义为对应的具体类型获取编译期校验；如果需要发送多种类型的消息，再定义为Object
• 泛型类型需要和配置的ProducerFactory的序列化器匹配：比如V定义为自定义的PlanDTO，那么value序列化器需要配置为支持该类型序列化的实现（如Jackson的JsonSerializer），否则运行时会抛出序列化异常
• send方法的入参类型需要和泛型定义匹配，否则编译阶段就会抛出类型不匹配的错误，提前规避问题。

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