springboot-kafka-module-skill

Spring Kafka Module Skill

面向已有 Spring Boot 项目的开发者，快速集成 Kafka 能力。

能力清单

能力	说明
生产者	同步/异步发送消息、消息分区、消息key
消费者	消费监听、消息重试、消费者组
消息序列化	JSON/Avro/ProtoBuf
事务消息	Kafka 事务
拦截器	消息发送/消费拦截
** Streams**	Kafka Streams 流处理

触发场景

用户说"帮我加 Kafka"或"集成 Kafka"时触发。

依赖配置

<!-- pom.xml 添加 -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

默认方法封装

1. 生产者

@Service
public class KafkaProducerService {
    
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    /**
     * 发送消息（同步）
     */
    public SendResult<String, String> send(String topic, String key, String message) {
        ListenableFuture<SendResult<String, String>> future = 
            kafkaTemplate.send(topic, key, message);
        try {
            return future.get(10, TimeUnit.SECONDS);
        } catch (Exception e) {
            throw new RuntimeException("发送消息失败", e);
        }
    }
    
    /**
     * 发送消息（异步）
     */
    public void sendAsync(String topic, String key, String message, SendCallback callback) {
        kafkaTemplate.send(topic, key, message).addCallback(callback);
    }
    
    /**
     * 发送 JSON 消息
     */
    public void sendJson(String topic, String key, Object object) {
        String json = JSON.toJSONString(object);
        kafkaTemplate.send(topic, key, json);
    }
    
    /**
     * 发送带回调的消息
     */
    public void sendWithCallback(String topic, String key, String message) {
        kafkaTemplate.send(topic, key, message).addCallback(
            success -> {
                // 成功
                String topic = success.getRecordMetadata().topic();
                int partition = success.getRecordMetadata().partition();
                long offset = success.getRecordMetadata().offset();
                log.info("消息发送成功: topic={}, partition={}, offset={}", topic, partition, offset);
            },
            failure -> {
                // 失败
                log.error("消息发送失败: {}", failure.getMessage());
            }
        );
    }
}