spring-kafka

Name: spring-kafka
Author: claude-dev-suite

github.com/claude-dev-suite开发4641 次安装2 天前更新

代码仓库

关于 spring-kafka

面向 Spring Kafka 开发者的快速参考技能，汇集生产者、消费者、错误处理与测试示例，并提供可靠性配置和故障排查要点。

主要功能

01KafkaTemplate 生产消息
02监听器与批量消费
03重试主题与死信处理
04EmbeddedKafka 和 Testcontainers 测试
05生产配置清单与故障排查

适用场景

01实现 Spring 应用的 Kafka 消息生产
02构建单条或批量消息消费者
03配置失败重试和死信主题
04测试 Kafka 集成并排查消费问题

MCP 相关

README 提及使用 `mcp__documentation__fetch_docs` 获取 Kafka 的深入文档。

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/claude-dev-suite/claude-dev-suite --skill spring-kafka

安装后在 Claude Code 中输入/spring-kafka确认已加载。

详细介绍

Spring Kafka - 快速参考

深入了解：使用技术为 kafka 的 mcp__documentation__fetch_docs 获取全面文档。

依赖项

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

配置

application.yml

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: my-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "com.example.dto"
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
      properties:
        enable.idempotence: true
    listener:
      ack-mode: manual
      concurrency: 3

生产者模式

KafkaTemplate

@Service
@RequiredArgsConstructor
public class OrderProducer {

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

    public void sendOrder(OrderEvent event) {
        kafkaTemplate.send("orders", event.getOrderId(), event)
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("Failed to send order: {}", event.getOrderId(), ex);
                } else {
                    log.info("Order sent: {} to partition {}",
                        event.getOrderId(),
                        result.getRecordMetadata().partition());
                }
            });
    }

    // With headers
    public void sendWithHeaders(OrderEvent event, String correlationId) {
        ProducerRecord<String, OrderEvent> record = new ProducerRecord<>(
            "orders", event.getOrderId(), event);
        record.headers()
            .add("correlation-id", correlationId.getBytes())
            .add("source", "order-service".getBytes());

        kafkaTemplate.send(record);
    }
}

事务型生产者

@Configuration
public class KafkaConfig {

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx-");
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        return new DefaultKafkaProducerFactory<>(config);
    }

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    @Bean
    public KafkaTransactionManager<String, Object> kafkaTransactionManager() {
        return new KafkaTransactionManager<>(producerFactory());
    }
}

@Service
@Transactional("kafkaTransactionManager")
public class TransactionalProducer {

    public void sendMultiple(List<OrderEvent> events) {
        events.forEach(e -> kafkaTemplate.send("orders", e.getId(), e));
    }
}

消费者模式

基本的 @KafkaListener

@Service
@RequiredArgsConstructor
public class OrderConsumer {

    @KafkaListener(topics = "orders", groupId = "order-processor")
    public void consume(
            @Payload OrderEvent event,
            @Header(KafkaHeaders.RECEIVED_KEY) String key,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            Acknowledgment ack) {

        log.info("Received order: {} from partition {} offset {}",
            event.getOrderId(), partition, offset);

        try {
            processOrder(event);
            ack.acknowledge();
        } catch (Exception e) {
            log.error("Failed to process order: {}", event.getOrderId(), e);
            throw e; // Will trigger retry
        }
    }
}

批量消费者

@KafkaListener(
    topics = "orders",
    groupId = "batch-processor",
    containerFactory = "batchKafkaListenerContainerFactory"
)
public void consumeBatch(
        List<OrderEvent> events,
        @Header(KafkaHeaders.RECEIVED_PARTITION) List<Integer> partitions,
        Acknowledgment ack) {

    log.info("Received batch of {} orders", events.size());
    processBatch(events);
    ack.acknowledge();
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, OrderEvent> batchKafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setBatchListener(true);
    factory.getContainerProperties().setAckMode(AckMode.MANUAL);
    return factory;
}

类级别监听器

@KafkaListener(topics = "orders", groupId = "order-handler")
@Service
public class OrderHandler {

    @KafkaHandler
    public void handleCreated(OrderCreatedEvent event) {
        // Handle order created
    }

    @KafkaHandler
    public void handleUpdated(OrderUpdatedEvent event) {
        // Handle order updated
    }

    @KafkaHandler(isDefault = true)
    public void handleDefault(Object event) {
        log.warn("Unknown event type: {}", event.getClass());
    }
}

重试主题（Spring Kafka 3.x）

@RetryableTopic

@RetryableTopic(
    attempts = "3",
    backoff = @Backoff(delay = 1000, multiplier = 2.0, maxDelay = 10000),
    dltStrategy = DltStrategy.FAIL_ON_ERROR,
    autoCreateTopics = "true",
    topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE
)
@KafkaListener(topics = "orders", groupId = "retry-consumer")
public void consumeWithRetry(OrderEvent event, Acknowledgment ack) {
    processOrder(event);
    ack.acknowledge();
}

@DltHandler
public void handleDlt(OrderEvent event,
        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
        @Header(KafkaHeaders.EXCEPTION_MESSAGE) String errorMessage) {

    log.error("DLT received: {} from {} - error: {}",
        event.getOrderId(), topic, errorMessage);
    // Store in database for manual review
    failedOrderRepository.save(new FailedOrder(event, errorMessage));
}

手动重试配置

@Configuration
@EnableKafka
public class KafkaRetryConfig {

    @Bean
    public RetryTopicConfiguration retryTopicConfiguration(KafkaTemplate<String, Object> template) {
        return RetryTopicConfigurationBuilder
            .newInstance()
            .maxAttempts(4)
            .fixedBackOff(3000)
            .includeTopic("orders")
            .doNotAutoCreateRetryTopics()
            .create(template);
    }
}

错误处理

自定义错误处理器

@Bean
public DefaultErrorHandler errorHandler(KafkaTemplate<String, Object> template) {
    // Send to DLT after 3 retries
    DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template,
        (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition()));

    DefaultErrorHandler handler = new DefaultErrorHandler(recoverer,
        new FixedBackOff(1000L, 3L));

    // Don't retry for these exceptions
    handler.addNotRetryableExceptions(
        ValidationException.class,
        DeserializationException.class
    );

    return handler;
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setCommonErrorHandler(errorHandler(kafkaTemplate()));
    return factory;
}

测试

@EmbeddedKafka

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = {"orders"})
class OrderProducerTest {

    @Autowired
    private EmbeddedKafkaBroker embeddedKafka;

    @Autowired
    private OrderProducer orderProducer;

    @Test
    void shouldSendOrder() throws Exception {
        OrderEvent event = new OrderEvent("123", "CREATED");

        Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(
            "test-group", "true", embeddedKafka);
        ConsumerFactory<String, OrderEvent> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
        Consumer<String, OrderEvent> consumer = cf.createConsumer();
        embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "orders");

        orderProducer.sendOrder(event);

        ConsumerRecord<String, OrderEvent> record = KafkaTestUtils.getSingleRecord(consumer, "orders");
        assertThat(record.value().getOrderId()).isEqualTo("123");
    }
}

使用 CountDownLatch 测试消费者

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = {"orders"})
class OrderConsumerTest {

    @SpyBean
    private OrderConsumer orderConsumer;

    @Autowired
    private EmbeddedKafkaBroker embeddedKafka;

    @Test
    void shouldConsumeAndProcessOrder() throws Exception {
        CountDownLatch latch = new CountDownLatch(1);
        doAnswer(inv -> { inv.callRealMethod(); latch.countDown(); return null; })
            .when(orderConsumer).consume(any(), any());

        Map<String, Object> props = KafkaTestUtils.producerProps(embeddedKafka);
        KafkaTemplate<String, String> template = new KafkaTemplate<>(
            new DefaultKafkaProducerFactory<>(props));
        template.send("orders", "{\"orderId\":\"456\",\"status\":\"CREATED\"}");

        assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
        verify(orderConsumer).consume(any(), any());
    }
}

Testcontainers

@SpringBootTest
@Testcontainers
class OrderIntegrationTest {

    @Container
    @ServiceConnection
    static KafkaContainer kafka = new KafkaContainer(
        DockerImageName.parse("apache/kafka-native:3.8.0"));

    @Autowired
    private KafkaTemplate<String, OrderEvent> kafkaTemplate;

    @Autowired
    private OrderRepository orderRepository;

    @Test
    void shouldProcessOrderEndToEnd() throws Exception {
        kafkaTemplate.send("orders", "key-1",
            new OrderEvent("789", "CREATED")).get(10, TimeUnit.SECONDS);

        await().atMost(Duration.ofSeconds(10))
            .untilAsserted(() -> {
                Optional<Order> order = orderRepository.findById("789");
                assertThat(order).isPresent();
                assertThat(order.get().getStatus()).isEqualTo("CREATED");
            });
    }
}

深入探究：有关 MockConsumer/MockProducer、@EmbeddedKafka 高级模式以及 Node.js/Python Kafka 测试，请参阅 messaging-testing-kafka 技能。

最佳实践

应该做	不要做
使用 `acks=all` 确保持久性	在生产环境中使用 `acks=0`
启用幂等性	忽略重复消息
为失败配置 DLT	静默丢弃失败消息
使用手动确认	未经处理就自动提交
设置正确的反序列化器信任配置	信任所有程序包

生产环境检查清单

不适合使用此技能的情况

原生 Kafka - 使用 kafka 技能配置 Broker
RabbitMQ - 改用 spring-amqp
简单消息传递 - 可考虑使用 Spring Events
Kafka Streams - 可能需要额外的技能

反模式

反模式	问题	解决方案
自动提交	消息丢失	使用手动确认
未配置错误处理器	静默失败	配置错误处理器
未配置 DLT	失败消息丢失	添加死信主题
在监听器中阻塞	消费者延迟	使用异步处理
反序列化器错误	消费时出错	与生产者序列化器匹配
未实现幂等性	重复处理	实现幂等消费者

快速故障排查

问题	诊断方法	修复方法
消费者未收到消息	检查 group.id	验证消费者组
序列化错误	检查值类型	配置正确的反序列化器
频繁重新平衡	检查 session.timeout	增加超时时间
消费者延迟	检查处理时间	优化或扩展消费者
消息进入 DLT	检查错误日志	修复处理错误

参考文档

支持 Markdown

登录后评论

基本信息

作者: claude-dev-suite
仓库: claude-dev-suite/claude-dev-suite
分类: 开发
安装次数: 41
GitHub Stars: 46

报告问题