kafka-patterns

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Apache Kafka 消息队列

来源：https://kafka.apache.org/documentation/
Spring Kafka：https://docs.spring.io/spring-kafka/reference/

Capability Boundaries

能力边界

✅ Strong Suits

✅ 适用场景

生产者配置 — acks(retry/partition)/retries(batchSize/linger)/幂等
消费者配置 — auto-offset-reset/enable-auto-commit/manual commit
幂等性保证 — 生产者 enable.idempotence + acks=all + 消费者业务去重
Spring Kafka — @KafkaListener / @RetryableTopic / @DltHandler
DLT死信 — DefaultErrorHandler + DeadLetterPublishingRecoverer
顺序消息 — 同一Key发送到同一分区
事务消息 — @Transactional + KafkaTransactionManager

生产者配置 — acks（重试/分区）/retries（批量大小/linger）/幂等性
消费者配置 — auto-offset-reset/enable-auto-commit/手动提交
幂等性保证 — 生产者 enable.idempotence + acks=all + 消费者业务去重
Spring Kafka — @KafkaListener / @RetryableTopic / @DltHandler
DLT 死信 — DefaultErrorHandler + DeadLetterPublishingRecoverer
顺序消息 — 同一 Key 发送到同一分区
事务消息 — @Transactional + KafkaTransactionManager

❌ Out of Scope

❌ 不在范围内

Kafka Streams → 流处理，不属于消息队列范畴
Kafka Connect → 数据管道，非 LLM 代码问题
Schema Registry → 序列化方案选型

Kafka Streams → 流处理，不属于消息队列范畴
Kafka Connect → 数据管道，非 LLM 代码问题
Schema Registry → 序列化方案选型

生产者 acks 选择决策

acks值	可靠性	延迟	适用场景
`acks=all` (-1)	⭐⭐⭐ 最高	高	核心业务(订单/支付/交易) ✅
`acks=1`	⭐⭐ 中	中	默认值，一般业务
`acks=0`	⭐ 最低	低	日志/监控/不重要的指标

核心业务必须:
acks=all
+
min.insync.replicas=2
(Broker配置) +
replication.factor=3

acks值	可靠性	延迟	适用场景
`acks=all` （-1）	⭐⭐⭐ 最高	高	核心业务（订单/支付/交易）✅
`acks=1`	⭐⭐ 中	中	默认值，一般业务
`acks=0`	⭐ 最低	低	日志/监控/不重要的指标

核心业务必须使用：
acks=all
+
min.insync.replicas=2
（Broker 配置）+
replication.factor=3

LLM最常犯的错误

LLM 最常犯的错误

#	错误	正确做法
1	acks=0 或 acks=1 丢消息	核心业务 acks=all + min.insync.replicas=2
2	enable.auto.commit=true（自动提交offset）	手动提交，处理完再 acknowledge
3	消费者不处理重复消费	业务层幂等(唯一键+数据库去重)
4	不配置DLT死信，异常消息无限重试	DefaultErrorHandler + DeadLetterPublishingRecoverer
5	所有消息发到同一个分区	用Key路由到分区，同一Key保证顺序
6	同步发送 `producer.send().get()`	异步发送 + callback 处理异常
7	不配置 enable.idempotence=true	幂等生产防止重试导致重复消息
8	监听器内异常不处理	配置 ErrorHandler，异常→DLT

#	错误	正确做法
1	acks=0 或 acks=1 丢失消息	核心业务使用 acks=all + min.insync.replicas=2
2	enable.auto.commit=true（自动提交 offset）	手动提交，处理完成后再 acknowledge
3	消费者未处理重复消费问题	业务层幂等（唯一键 + 数据库去重）
4	未配置 DLT 死信，异常消息无限重试	DefaultErrorHandler + DeadLetterPublishingRecoverer
5	所有消息发到同一个分区	使用 Key 路由到分区，同一 Key 保证顺序
6	同步发送 `producer.send().get()`	异步发送 + callback 处理异常
7	未配置 enable.idempotence=true	幂等生产，防止重试导致消息重复
8	监听器内未处理异常	配置 ErrorHandler，异常 → DLT

核心模式

模式 1: 生产者配置（核心业务）

模式 1：生产者配置（核心业务）

java

@Bean
public ProducerFactory<String, Object> producerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    config.put(ProducerConfig.ACKS_CONFIG, "all");                    // ✅ 所有副本确认
    config.put(ProducerConfig.RETRIES_CONFIG, 3);                      // 重试3次
    config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);        // ✅ 幂等
    config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
    config.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);    // 投递超时
    config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "zstd");        // 压缩
    config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
    return new DefaultKafkaProducerFactory<>(config);
}

java

@Bean
public ProducerFactory<String, Object> producerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    config.put(ProducerConfig.ACKS_CONFIG, "all");                    // ✅ 所有副本确认
    config.put(ProducerConfig.RETRIES_CONFIG, 3);                      // 重试3次
    config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);        // ✅ 幂等
    config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
    config.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);    // 投递超时
    config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "zstd");        // 压缩
    config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
    return new DefaultKafkaProducerFactory<>(config);
}

模式 2: 消费者手动提交 + DLT

模式 2：消费者手动提交 + DLT

java

@Bean
public ConsumerFactory<String, Object> consumerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ConsumerConfig.GROUP_ID_CONFIG, "order-group");
    config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");   // 从最早未消费开始
    config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);       // ✅ 手动提交
    config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
    config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    return new DefaultKafkaConsumerFactory<>(config);
}

// ✅ DLT 配置
@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory(
        ConsumerFactory<String, Object> cf, KafkaTemplate<String, Object> kt) {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(cf);
    // 重试3次，每次间隔1秒，之后进DLT
    factory.setCommonErrorHandler(new DefaultErrorHandler(
        new DeadLetterPublishingRecoverer(kt),
        new FixedBackOff(1000L, 3L)));
    return factory;
}

java

@Bean
public ConsumerFactory<String, Object> consumerFactory() {
    Map<String, Object> config = new HashMap<>();
    config.put(ConsumerConfig.GROUP_ID_CONFIG, "order-group");
    config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");   // 从最早未消费开始
    config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);       // ✅ 手动提交
    config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
    config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    return new DefaultKafkaConsumerFactory<>(config);
}

// ✅ DLT 配置
@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory(
        ConsumerFactory<String, Object> cf, KafkaTemplate<String, Object> kt) {
    ConcurrentKafkaListenerContainerFactory<String, Object> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(cf);
    // 重试3次，每次间隔1秒，之后进DLT
    factory.setCommonErrorHandler(new DefaultErrorHandler(
        new DeadLetterPublishingRecoverer(kt),
        new FixedBackOff(1000L, 3L)));
    return factory;
}

模式 3: Spring Kafka 监听器 + DLT

模式 3：Spring Kafka 监听器 + DLT

java

// 方式一：直接 @KafkaListener + DLT ErrorHandler
@KafkaListener(topics = "order-topic", groupId = "order-group")
public void listen(ConsumerRecord<String, OrderEvent> record, Acknowledgment ack) {
    try {
        processOrder(record.value());
        ack.acknowledge();          // ✅ 手动提交 offset
    } catch (Exception e) {
        log.error("处理订单失败: key={}", record.key(), e);
        throw e;                    // 抛出 → ErrorHandler → 重试 → DLT
    }
}

// 方式二：@RetryableTopic 自动重试 + DLT
@RetryableTopic(
    attempts = "4",
    backoff = @Backoff(delay = 1000, multiplier = 2.0, maxDelay = 10000),
    kafkaTemplate = "kafkaTemplate")
@KafkaListener(topics = "order-topic", groupId = "order-group")
public void processOrder(OrderEvent event) {
    // 业务逻辑，失败自动重试(1s→2s→4s→...)
}

@DltHandler
public void handleDlt(OrderEvent event) {
    log.error("消息进入死信队列: {}", event);
    // 人工处理或告警
}

java

// 方式一：直接 @KafkaListener + DLT ErrorHandler
@KafkaListener(topics = "order-topic", groupId = "order-group")
public void listen(ConsumerRecord<String, OrderEvent> record, Acknowledgment ack) {
    try {
        processOrder(record.value());
        ack.acknowledge();          // ✅ 手动提交 offset
    } catch (Exception e) {
        log.error("处理订单失败: key={}", record.key(), e);
        throw e;                    // 抛出 → ErrorHandler → 重试 → DLT
    }
}

// 方式二：@RetryableTopic 自动重试 + DLT
@RetryableTopic(
    attempts = "4",
    backoff = @Backoff(delay = 1000, multiplier = 2.0, maxDelay = 10000),
    kafkaTemplate = "kafkaTemplate")
@KafkaListener(topics = "order-topic", groupId = "order-group")
public void processOrder(OrderEvent event) {
    // 业务逻辑，失败自动重试(1s→2s→4s→...)
}

@DltHandler
public void handleDlt(OrderEvent event) {
    log.error("消息进入死信队列: {}", event);
    // 人工处理或告警
}

模式 4: 消息发送（异步 + callback）

模式 4：消息发送（异步 + callback）

java

@Service
public class OrderEventPublisher {
    private final KafkaTemplate<String, Object> kafkaTemplate;

    public void publishOrderCreated(OrderEvent event) {
        String key = String.valueOf(event.getOrderId());  // ✅ 同一订单→同一分区
        ListenableFuture<SendResult<String, Object>> future =
            kafkaTemplate.send("order-topic", key, event);

        // ✅ 异步处理发送结果
        future.addCallback(result -> log.debug("发送成功: {}", result.getRecordMetadata().offset()),
            ex -> log.error("发送失败: key={}", key, ex));
    }
}

java

@Service
public class OrderEventPublisher {
    private final KafkaTemplate<String, Object> kafkaTemplate;

    public void publishOrderCreated(OrderEvent event) {
        String key = String.valueOf(event.getOrderId());  // ✅ 同一订单→同一分区
        ListenableFuture<SendResult<String, Object>> future =
            kafkaTemplate.send("order-topic", key, event);

        // ✅ 异步处理发送结果
        future.addCallback(result -> log.debug("发送成功: {}", result.getRecordMetadata().offset()),
            ex -> log.error("发送失败: key={}", key, ex));
    }
}

模式 5: 业务幂等(消费者端去重)

模式 5：业务幂等（消费者端去重）

java

@KafkaListener(topics = "order-topic")
public void listen(OrderEvent event) {
    // ✅ 业务幂等：用事件ID去重
    String eventId = event.getEventId();
    Boolean existed = redisTemplate.opsForValue().setIfAbsent(
        "kafka:dedup:" + eventId, "1", Duration.ofHours(24));
    if (Boolean.FALSE.equals(existed)) {
        log.debug("重复事件跳过: {}", eventId);
        return;
    }
    // 处理业务
    processOrder(event);
}

java

@KafkaListener(topics = "order-topic")
public void listen(OrderEvent event) {
    // ✅ 业务幂等：用事件ID去重
    String eventId = event.getEventId();
    Boolean existed = redisTemplate.opsForValue().setIfAbsent(
        "kafka:dedup:" + eventId, "1", Duration.ofHours(24));
    if (Boolean.FALSE.equals(existed)) {
        log.debug("重复事件跳过: {}", eventId);
        return;
    }
    // 处理业务
    processOrder(event);
}

Gotchas

常见陷阱

acks=all 最安全但最慢 — 日志/监控类消息可用 acks=1
幂等性 enable.idempotence=true 要求 acks=all + max.in.flight.requests.per.connection<=5
手动提交 offset 必须 finally 或 Spring Acknowledgment — 漏提交 = 重复消费
DLT 重试次数有限 — 超过重试次数的消息进入 DLT Topic，需要人工处理
同一 group 内每个分区只被一个消费者消费 — 消费者数 > 分区数 → 多余消费者空闲
offset 从 earliest 开始时重放所有历史消息 — 新消费者组需要注意
生产者和消费者的序列化器要匹配 — 用 JsonSerializer/JsonDeserializer 替代 String
异常重试时可能乱序 — 严格顺序场景用单分区或 RabbitMQ
@RetryableTopic 会创建额外 retry topic — 注意 Topic 数量膨胀
压缩(zstd/snappy) 降低网络带宽 — 推荐 zstd(压缩比高)
max.poll.records 控制每批拉取数量 — 匹配处理速度，避免 rebalance
大消息(>1MB)需要配置 max.message.bytes — 建议 < 10MB

acks=all 最安全但最慢 — 日志/监控类消息可用 acks=1
幂等性 enable.idempotence=true 要求 acks=all + max.in.flight.requests.per.connection<=5
手动提交 offset 必须使用 finally 或 Spring Acknowledgment — 漏提交 = 重复消费
DLT 重试次数有限 — 超过重试次数的消息进入 DLT Topic，需要人工处理
同一 group 内每个分区只被一个消费者消费 — 消费者数 > 分区数 → 多余消费者空闲
offset 从 earliest 开始时会重放所有历史消息 — 新消费者组需要注意
生产者和消费者的序列化器要匹配 — 使用 JsonSerializer/JsonDeserializer 替代 String
异常重试时可能乱序 — 严格顺序场景使用单分区或 RabbitMQ
@RetryableTopic 会创建额外的 retry topic — 注意 Topic 数量膨胀
压缩（zstd/snappy）可降低网络带宽 — 推荐 zstd（压缩比高）
max.poll.records 控制每批拉取数量 — 匹配处理速度，避免 rebalance
大消息（>1MB）需要配置 max.message.bytes — 建议 < 10MB

Data Privacy

数据隐私

本技能不收集、存储或传输任何用户数据。