【Kafka】项目整合使用案例

【一】需求描述

【1】避免数据丢失

数据丢失避免需要从生产者和消费者两端配置。生产者必须确保消息成功提交到Kafka（acks=all），并启用重试机制。

消费者则需要手动提交偏移量，并在处理完成后才提交，避免消息丢失但可能重复消费（因此消费者逻辑要幂等）。

【2】封装工具

将生产者和消费者的通用操作抽象出来。

【3】主题枚举

主题枚举有助于管理主题名称，避免硬编码。

【4】消费失败处理

消费失败处理通常采用重试机制，比如Spring Kafka的RetryTemplate，但要注意无限重试可能阻塞消费，所以需要设置最大重试次数并最终转移到死信队列（DLQ）进行人工干预。

【5】顺序消费

顺序消费是Kafka的难点，因为默认并行消费会乱序。需要利用Kafka的分区顺序性，确保同一关键字的消息发送到同一分区，然后单个消费者线程处理该分区。多线程顺序消费则需要在分区内串行，但可以多个分区并行——即每个分区一个线程，内部单线程处理。

【6】多线程顺序消费

根据消息关键字路由到不同线程，每个线程独立处理一批消息。

【二】实现过程

【1】yml配置

  kafka:bootstrap-servers: localhost:9092producer:# 发生错误后，消息重发的次数。retries: 3#当有多个消息需要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算。batch-size: 16384# 设置生产者内存缓冲区的大小。buffer-memory: 33554432# 键的序列化方式key-serializer: org.apache.kafka.common.serialization.StringSerializer# 值的序列化方式value-serializer: org.apache.kafka.common.serialization.StringSerializer# acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。# acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。# acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。acks: 1consumer:# 自动提交的时间间隔 在spring boot 2.X 版本中这里采用的是值的类型为Duration 需要符合特定的格式，如1S,1M,2H,5Dauto-commit-interval: 1S# 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：# latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）# earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录auto-offset-reset: earliest# 是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量enable-auto-commit: false# 键的反序列化方式key-deserializer: org.apache.kafka.common.serialization.StringDeserializer# 值的反序列化方式value-deserializer: org.apache.kafka.common.serialization.StringDeserializerlistener:# 在侦听器容器中运行的线程数。concurrency: 5#listner负责ack，每调用一次，就立即commitack-mode: manual_immediatemissing-topics-fatal: false

【2】Java配置类：KafkaConfig

package com.allen.study.common.config;import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.*;import java.util.HashMap;
import java.util.Map;/*** @ClassName: KafkaConfig* @Author: AllenSun* @Date: 2025/8/27 12:57*/
@Configuration
@EnableKafka
public class KafkaConfig {@Value("${spring.kafka.bootstrap-servers}")private String bootstrapServers;@Beanpublic ProducerFactory<String, String> producerFactory() {Map<String, Object> configProps = new HashMap<>();configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);configProps.put(ProducerConfig.ACKS_CONFIG, "all");configProps.put(ProducerConfig.RETRIES_CONFIG, 3);configProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);configProps.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);return new DefaultKafkaProducerFactory<>(configProps);}@Beanpublic KafkaTemplate<String, String> kafkaTemplate() {return new KafkaTemplate<>(producerFactory());}@Beanpublic ConsumerFactory<String, String> consumerFactory() {Map<String, Object> configProps = new HashMap<>();configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);configProps.put(ConsumerConfig.GROUP_ID_CONFIG, "your-group-id");configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);configProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);configProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");configProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);return new DefaultKafkaConsumerFactory<>(configProps);}
}

【3】主题枚举：KafkaTopicEnum

package com.allen.study.common.constant;import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;public enum KafkaTopicEnum {ADD_EMP("add_emp", 3, 3),EDIT_EMP("edit_emp", 3, 3),DEL_EMP("del_emp", 3, 3),ADD_PRODUCT("add_product", 3, 3),EDIT_PRODUCT("edit_product", 3, 3),DEL_PRODUCT("del_product", 3, 3);private final String topicName;private final int partitions;private final int replicationFactor;KafkaTopicEnum(String topicName, int partitions, int replicationFactor) {this.topicName = topicName;this.partitions = partitions;this.replicationFactor = replicationFactor;}public String getTopicName() {return topicName;}public int getPartitions() {return partitions;}public int getReplicationFactor() {return replicationFactor;}public static List<String> getAllTopics() {return Arrays.stream(values()).map(KafkaTopicEnum::getTopicName).collect(Collectors.toList());}
}

【4】生产者工具封装：KafkaProducerUtil

（1）工具代码

package com.allen.study.common.utils.kafka;import com.allen.study.common.constant.KafkaTopicEnum;
import com.allen.study.common.exception.CustomRuntimeException;
import com.allen.study.common.exception.KafkaException;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;import java.util.concurrent.TimeUnit;/*** @ClassName: KafkaUtils* @Author: AllenSun* @Date: 2025/3/5 17:13*/
@Component
@Slf4j
public class KafkaProducerUtil {private final KafkaTemplate<String, String> kafkaTemplate;private final ObjectMapper objectMapper;public KafkaProducerUtil(KafkaTemplate<String, String> kafkaTemplate, ObjectMapper objectMapper) {this.kafkaTemplate = kafkaTemplate;this.objectMapper = objectMapper;}/*** 发送消息（同步方式，保证可靠性）*/public <T> boolean sendMessage(KafkaTopicEnum topic, String key, T message) {return sendMessage(topic, key, message, 0);}/*** 发送消息（支持重试）*/public <T> boolean sendMessage(KafkaTopicEnum topic, String key, T message, int retryCount) {String messageJson;try {messageJson = objectMapper.writeValueAsString(message);} catch (JsonProcessingException e) {log.error("消息序列化失败: {}", message, e);throw new KafkaException("消息序列化失败");}log.debug("消息序列化成功：{}", messageJson);int attempt = 0;while (attempt <= retryCount) {try {ListenableFuture<SendResult<String, String>> future =kafkaTemplate.send(topic.getTopicName(), key, messageJson);// 同步等待发送结果SendResult<String, String> result = future.get(10, TimeUnit.SECONDS);log.debug("消息发送成功: topic={}, key={}, partition={}, offset={}, content={}",topic.getTopicName(), key,result.getRecordMetadata().partition(),result.getRecordMetadata().offset(),messageJson);return true;} catch (Exception e) {attempt++;log.warn("消息发送失败(尝试 {}): topic={}, key={}",attempt, topic.getTopicName(), key, e);if (attempt > retryCount) {log.error("重试次数耗尽，消息发送最终失败: topic={}, key={}, message={}",topic.getTopicName(), key, messageJson, e);// 这里可以加入死信队列或落库重试机制// 发送到死信队列kafkaTemplate.send("dead.letter.producer." + topic.getTopicName(), key, messageJson);return false;}try {Thread.sleep(1000 * attempt); // 指数退避} catch (InterruptedException ie) {Thread.currentThread().interrupt();return false;}}}return false;}/*** 发送消息（异步方式，高性能）*/public <T> void sendMessageAsync(KafkaTopicEnum topic, String key, T message) {String messageJson;try {messageJson = objectMapper.writeValueAsString(message);} catch (JsonProcessingException e) {log.error("消息序列化失败: {}", message, e);return;}log.debug("消息序列化成功：{}", messageJson);ListenableFuture<SendResult<String, String>> future =kafkaTemplate.send(topic.getTopicName(), key, messageJson);future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {@Overridepublic void onSuccess(SendResult<String, String> result) {log.debug("异步消息发送成功: topic={}, key={}, partition={}, offset={}, content={}",topic.getTopicName(), key,result.getRecordMetadata().partition(),result.getRecordMetadata().offset(),messageJson);}@Overridepublic void onFailure(Throwable ex) {log.error("异步消息发送失败: topic={}, key={}, message={}",topic.getTopicName(), key, messageJson, ex);// 异步发送失败处理逻辑// throw new KafkaException("异步消息发送失败：{}"+ex.getMessage());}});}/*** 发送消息到指定主题** @param topic   主题名称* @param message 消息内容*/public void sendMessage(String topic, String message) {ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {@Overridepublic void onFailure(Throwable ex) {throw new CustomRuntimeException("kafka消息发送失败：{}"+ex.getMessage());}@Overridepublic void onSuccess(SendResult<String, String> result) {log.info("kafka消息发送成功: " + message + ", partition: " + result.getRecordMetadata().partition() + ", " +"offset: " + result.getRecordMetadata().offset());}});}
}

（2）生产者实例

    /*** 创建商品信息表** @param productInfo 商品信息表实体* @return 响应结果*/public DomainResponse<Void> create(@NotNull ProductInfo productInfo) {// 保存商品信息表productInfo.setId(IdUtil.getSnowflakeNextIdStr());productInfoRepo.create(productInfo);kafkaProducerUtil.sendMessageAsync(KafkaTopicEnum.ADD_PRODUCT,productInfo.getId(),productInfo);return DomainResponse.ok();}

【5】消费者工具类与异常处理：BaseKafkaConsumer

（1）工具代码

package com.allen.study.common.utils.kafka;import com.allen.study.common.exception.RetryableException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;/*** @ClassName: BaseKafkaConsumer* @Author: AllenSun* @Date: 2025/8/27 13:16*/
@Slf4j
public abstract class BaseKafkaConsumer<T> {private final ObjectMapper objectMapper;private final Class<T> messageType;private KafkaTemplate<String,String> kafkaTemplate;public BaseKafkaConsumer(ObjectMapper objectMapper, Class<T> messageType) {this.objectMapper = objectMapper;this.messageType = messageType;}/*** 消息处理模板方法* 该方法实现了消息消费的基本流程，包括反序列化、业务处理和异常处理* @param record 接收到的消息记录，包含主题、分区、偏移量等信息* @param ack 消息确认对象，用于确认消息是否成功处理*/protected void processMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {// 用于存储反序列化后的消息对象T message;try {// 使用ObjectMapper将消息字符串反序列化为指定类型的对象message = objectMapper.readValue(record.value(), messageType);} catch (Exception e) {// 反序列化失败时的处理：记录错误日志并确认消费log.error("消息反序列化失败: topic={}, partition={}, offset={}, value={}",record.topic(), record.partition(), record.offset(), record.value(), e);// 确认消费（避免死信）ack.acknowledge();// 结束当前消息处理return;}try {// 调用具体业务处理方法，处理反序列化后的消息handleMessage(message, record.key(), record);// 处理成功，确认消费ack.acknowledge();// 记录成功处理日志log.debug("消息处理成功: topic={}, key={}, partition={}, offset={}",record.topic(), record.key(), record.partition(), record.offset());log.debug("消息内容：content={}",record.value());} catch (RetryableException e) {// 可重试异常：记录警告日志但不确认消费，等待重试机制触发log.warn("消息处理失败，需要重试: topic={}, key={}", record.topic(), record.key(), e);log.warn("消息内容：content={}",record.value());// 不确认消费，等待重试} catch (Exception e) {// 其他异常：记录错误日志并确认消费，避免消息进入死信队列log.error("消息处理失败: topic={}, key={}, message={}",record.topic(), record.key(), record.value(), e);log.error("消息内容：content={}",record.value());// ack.acknowledge(); // 确认消费（避免死信）// 可以发送到死信队列或记录错误日志// 发送到死信队列kafkaTemplate.send("dead.letter.consumer." + record.topic(), record.key(), record.value());}}/*** 抽象方法，由具体业务实现*/protected abstract void handleMessage(T message, String key, ConsumerRecord<String, String> record);
}

（2）消费者实例

package com.allen.study.application.elasticSearch.esSyncMq;import com.allen.study.application.elasticSearch.es_assembler.ProductInfoEsEntityAssembler;
import com.allen.study.application.elasticSearch.es_entity.ProductES;
import com.allen.study.application.elasticSearch.es_service.ProductSearchService;
import com.allen.study.common.utils.kafka.BaseKafkaConsumer;
import com.allen.study.domain.entity.ProductInfo;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;@Component
@Slf4j
public class AddProductListener extends BaseKafkaConsumer<ProductInfo> {private final ProductSearchService productSearchService;private final ProductInfoEsEntityAssembler assembler;public AddProductListener(ObjectMapper objectMapper, ProductSearchService productSearchService,ProductInfoEsEntityAssembler assembler) {super(objectMapper, ProductInfo.class);this.productSearchService = productSearchService;this.assembler = assembler;}@KafkaListener(topics = "add_product", groupId = "product_info", containerFactory ="orderContainerFactory")public void onMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {processMessage(record,ack);}@Overrideprotected void handleMessage(ProductInfo message, String key, ConsumerRecord<String, String> record) {// 1. 转化对象（或者你也可以重写Serializer<T>）ProductES productES = assembler.info2EsEntity(message);// 2. 处理新建后的一系列操作productSearchService.saveDocument(productES);log.info("业务处理完成");}
}

【6】顺序消费实现

（1）单线程顺序消费

1-配置工厂

package com.allen.study.common.utils.kafka;import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;/*** @ClassName: OrderKafkaConfig* @Author: AllenSun* @Date: 2025/8/27 16:18*/
@Configuration
public class OrderKafkaConfig {@Beanpublic ConcurrentKafkaListenerContainerFactory<String, String> orderContainerFactory(ConsumerFactory<String, String> consumerFactory) {ConcurrentKafkaListenerContainerFactory<String, String> factory =new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory);// 关键配置：单线程消费，保证顺序factory.setConcurrency(1);// 手动提交偏移量factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);// 设置异常处理器factory.setErrorHandler(new SeekToCurrentErrorHandler());return factory;}
}

2-使用案例

    @KafkaListener(topics = "add_product", groupId = "product_info", containerFactory ="orderContainerFactory")public void onMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {processMessage(record,ack);}

（2）多线程顺序消费

1-配置工厂

package com.allen.study.common.utils.kafka;import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;import java.util.HashMap;
import java.util.Map;/*** @ClassName: PartitionOrderKafkaConfig* @Author: AllenSun* @Date: 2025/8/27 16:22*/
@Configuration
public class PartitionOrderKafkaConfig {@Beanpublic ConcurrentKafkaListenerContainerFactory<String, String> partitionOrderContainerFactory(ConsumerFactory<String, String> consumerFactory) {ConcurrentKafkaListenerContainerFactory<String, String> factory =new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory);// 为每个分区创建单独的消费者实例factory.setConcurrency(3); // 与分区数匹配// 按key进行消费路由，确保相同key的消息由同一线程处理factory.setConsumerFactory(configureConsumerFactory(consumerFactory));factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);return factory;}private ConsumerFactory<String, String> configureConsumerFactory(ConsumerFactory<String, String> originalFactory) {Map<String, Object> configs = new HashMap<>(originalFactory.getConfigurationProperties());// 确保相同key的消息路由到相同分区configs.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"org.apache.kafka.clients.producer.UniformStickyPartitioner");return new DefaultKafkaConsumerFactory<>(configs);}
}

2-使用案例

    @KafkaListener(topics = "del_product", groupId = "product_info", containerFactory = "partitionOrderContainerFactory")public void onMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {String key = record.key();// 1. 获取分布式锁String DEL_PRODUCT_MQ_LOCK = "delproduct:lock:" + key;RLock lockRLock = redisLock.getRLock(DEL_PRODUCT_MQ_LOCK);boolean tryLock = lockRLock.tryLock();if(tryLock){try {processMessage(record,ack);} catch (Exception e) {log.error("分区顺序消费处理失败: key={}, offset={}", key, record.offset(), e);} finally {lockRLock.unlock();}}}

【7】死信队列与重试机制

（1）配置工厂

package com.allen.study.common.utils.kafka;import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.retry.backoff.FixedBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;/*** @ClassName: RetryKafkaConfig* @Author: AllenSun* @Date: 2025/8/27 16:24*/
@Configuration
@Slf4j
@AllArgsConstructor
public class RetryKafkaConfig {private final KafkaTemplate kafkaTemplate;@Beanpublic ConcurrentKafkaListenerContainerFactory<String, String> retryContainerFactory(ConsumerFactory<String, String> consumerFactory) {ConcurrentKafkaListenerContainerFactory<String, String> factory =new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory);// 配置重试机制RetryTemplate retryTemplate = new RetryTemplate();FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();backOffPolicy.setBackOffPeriod(1000); // 重试间隔1秒retryTemplate.setBackOffPolicy(backOffPolicy);retryTemplate.setRetryPolicy(new SimpleRetryPolicy(3)); // 最大重试3次factory.setRetryTemplate(retryTemplate);// 重试耗尽后处理factory.setRecoveryCallback(context -> {ConsumerRecord<?, ?> record = (ConsumerRecord<?, ?>) context.getAttribute("record");log.error("消息重试耗尽: topic={}, key={}, offset={}",record.topic(), record.key(), record.offset());// 发送到死信队列kafkaTemplate.send("dead.letter." + record.topic(),record.key().toString(),record.value().toString());return null;});return factory;}
}

（2）使用案例

    @KafkaListener(topics = "edit_product", groupId = "product_info", containerFactory = "retryContainerFactory")public void onMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {processMessage(record,ack);}

【8】@RetryableTopic和@DltHandler注解（Spring Kafka 2.7+）

（1）介绍

使用Spring Kafka的@RetryableTopic和@DltHandler注解来实现重试和死信队列功能。这两个注解是Spring Kafka 2.7版本引入的，用于简化重试和死信队列的配置。

（2）生产者案例

@Component
@Slf4j
@RequiredArgsConstructor
public class OrderEventConsumer {private final OrderService orderService;/*** 使用 @RetryableTopic 注解配置重试机制* attempts: 重试次数（包括初始尝试）* backoff: 退避策略配置* timeout: 超时时间* include: 指定哪些异常需要重试* exclude: 指定哪些异常不需要重试* autoCreateTopics: 是否自动创建重试和死信主题* topicSuffixingStrategy: 主题后缀策略*/@RetryableTopic(attempts = "4", // 总共尝试4次（初始1次 + 重试3次）backoff = @Backoff(delay = 1000,        // 初始延迟1秒multiplier = 2.0,    // 延迟倍数maxDelay = 10000     // 最大延迟10秒),timeout = "30000",       // 超时时间30秒include = {RetryableException.class},exclude = {NonRetryableException.class},autoCreateTopics = "true",topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE)@KafkaListener(topics = "order.process", groupId = "order-process-group")public void consumeOrderEvent(OrderEvent event) {log.info("收到订单事件: orderId={}, status={}", event.getOrderId(), event.getStatus());try {// 处理订单业务逻辑orderService.processOrder(event);log.info("订单处理成功: orderId={}", event.getOrderId());} catch (NonRetryableException e) {log.error("不可重试异常，消息将不进行重试: orderId={}", event.getOrderId(), e);throw e; // 抛出非重试异常，直接进入死信队列} catch (Exception e) {log.warn("订单处理失败，将进行重试: orderId={}", event.getOrderId(), e);throw new RetryableException("订单处理失败，需要重试", e);}}/*** 死信队列处理器 - 处理所有重试耗尽的消息* 可以针对不同主题设置不同的死信处理器*/@DltHandlerpublic void handleDlt(OrderEvent event) {log.error("消息进入死信队列: orderId={}, status={}", event.getOrderId(), event.getStatus());// 死信处理逻辑：记录日志、发送警报、人工干预等orderService.handleFailedOrder(event);// 可以在这里将死信消息存储到数据库或发送到其他系统log.warn("死信消息已处理: orderId={}", event.getOrderId());}
}

注意：上面的代码中，我们使用@RetryableTopic注解配置了重试主题，重试4次（总共5次消费尝试），重试间隔初始1秒，每次重试间隔乘以2。重试主题的后缀策略是使用索引值（默认会生成主题：原始主题名-retry-0, -retry-1等）。我们指定只有MyRetryableException异常才会重试。

如果重试次数用尽，消息会被发送到死信主题（默认是原始主题名-dlt）。然后通过@DltHandler方法来处理死信消息。

另外，我们设置了autoCreateTopics为false，这意味着我们需要提前创建好主题。如果你希望自动创建主题，可以设置为true，但需要确保Kafka broker允许自动创建主题。

需要创建的主题包括：
（1）原始主题：my-topic
（2）重试主题：my-topic-retry-0, my-topic-retry-1, my-topic-retry-2（因为重试次数为4，所以有3个重试主题，注意：重试次数包括第一次尝试，所以重试主题数量为attempts-1）
（3）死信主题：my-topic-dlt

关于重试主题的命名和数量，Spring Kafka的@RetryableTopic会根据重试次数和TopicSuffixingStrategy来生成。上述配置中，我们使用SUFFIX_WITH_INDEX_VALUE，所以重试主题会以索引值作为后缀。

另外，需要注意的是，重试主题的分区数默认与原始主题相同，我们也可以通过numPartitions属性来指定。

（3）工作原理说明

（1）@RetryableTopic 工作原理
1-当消费者方法抛出配置的重试异常时，Spring Kafka 会自动将消息发送到重试主题
2-重试主题的命名格式为：原始主题 + 重试后缀（如：order.process-retry-0, order.process-retry-1）
3-消息会在配置的延迟时间后重新投递到主主题
4-经过所有重试尝试后，如果仍然失败，消息会被发送到死信主题（order.process-dlt）

（2）@DltHandler 工作原理
1-@DltHandler注解的方法专门处理死信主题中的消息
2-当重试耗尽后，消息会自动进入死信主题
3-死信处理器可以针对不同类型的消息实现不同的处理逻辑

（3）主题自动创建
当 autoCreateTopics设置为 true时，Spring Kafka 会自动创建：
1-主主题：order.process
2-重试主题：order.process-retry-0, order.process-retry-1, order.process-retry-2
3-死信主题：order.process-dlt

【三】方案总结

【1】保证消息的消费顺序

在使用消息队列的过程中经常有业务场景需要严格保证消息的消费顺序，比如我们同时发了 2 个消息，这 2 个消息对应的操作分别对应的数据库操作是：

（1）更改用户会员等级。
（2）根据会员等级计算订单价格。

假如这两条消息的消费顺序不一样造成的最终结果就会截然不同。

Kafka 中 Partition(分区)是真正保存消息的地方，我们发送的消息都被放在了这里。而我们的 Partition(分区) 又存在于 Topic(主题) 这个概念中，并且我们可以给特定 Topic 指定多个 Partition。

每次添加消息到 Partition(分区) 的时候都会采用尾加法，如上图所示。 Kafka 只能为我们保证 Partition(分区) 中的消息有序。

消息在被追加到 Partition(分区)的时候都会分配一个特定的偏移量（offset）。Kafka 通过偏移量（offset）来保证消息在分区内的顺序性。

所以，我们就有一种很简单的保证消息消费顺序的方法：1 个 Topic 只对应一个 Partition。这样当然可以解决问题，但是破坏了 Kafka 的设计初衷。

Kafka 中发送 1 条消息的时候，可以指定 topic, partition, key,data（数据） 4 个参数。如果你发送消息的时候指定了 Partition 的话，所有消息都会被发送到指定的 Partition。并且，同一个 key 的消息可以保证只发送到同一个 partition，这个我们可以采用表/对象的 id 来作为 key 。

（1）1 个 Topic 只对应一个 Partition。
（2）（推荐）发送消息的时候指定 key/Partition。

【2】保证消息不丢失

（1）生产者丢失消息

生产者(Producer) 调用send方法发送消息之后，消息可能因为网络问题并没有发送过去。

所以，我们不能默认在调用send方法发送消息之后消息发送成功了。为了确定消息是发送成功，我们要判断消息发送的结果。但是要注意的是 Kafka 生产者(Producer) 使用 send 方法发送消息实际上是异步的操作，我们可以通过 get()方法获取调用结果，但是这样也让它变为了同步操作，示例代码如下：

SendResult<String, Object> sendResult = kafkaTemplate.send(topic, o).get();
if (sendResult.getRecordMetadata() != null) {logger.info("生产者成功发送消息到" + sendResult.getProducerRecord().topic() + "-> " + sendResult.getProducerRecord().value().toString());
}

但是一般不推荐这么做！可以采用为其添加回调函数的形式，示例代码如下：

        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, o);future.addCallback(result -> logger.info("生产者成功发送消息到topic:{} partition:{}的消息", result.getRecordMetadata().topic(), result.getRecordMetadata().partition()),ex -> logger.error("生产者发送消失败，原因：{}", ex.getMessage()));

如果消息发送失败的话，我们检查失败的原因之后重新发送即可！

另外，这里推荐为 Producer 的retries（重试次数）设置一个比较合理的值，一般是 3 ，但是为了保证消息不丢失的话一般会设置比较大一点。设置完成之后，当出现网络问题之后能够自动重试消息发送，避免消息丢失。另外，建议还要设置重试间隔，因为间隔太小的话重试的效果就不明显了，网络波动一次你 3 次一下子就重试完了。

（2）消费者丢失消息

消息在被追加到 Partition(分区)的时候都会分配一个特定的偏移量（offset）。偏移量（offset)表示 Consumer 当前消费到的 Partition(分区)的所在的位置。Kafka 通过偏移量（offset）可以保证消息在分区内的顺序性。

当消费者拉取到了分区的某个消息之后，消费者会自动提交了 offset。自动提交的话会有一个问题，试想一下，当消费者刚拿到这个消息准备进行真正消费的时候，突然挂掉了，消息实际上并没有被消费，但是 offset 却被自动提交了。

解决办法也比较粗暴，我们手动关闭自动提交 offset，每次在真正消费完消息之后再自己手动提交 offset 。 但是，这样会带来消息被重新消费的问题。比如你刚刚消费完消息之后，还没提交 offset，结果自己挂掉了，那么这个消息理论上就会被消费两次。

（3）Kafka丢失消息

Kafka 为分区（Partition）引入了多副本（Replica）机制。分区（Partition）中的多个副本之间会有一个叫做 leader 的家伙，其他副本称为 follower。我们发送的消息会被发送到 leader 副本，然后 follower 副本才能从 leader 副本中拉取消息进行同步。生产者和消费者只与 leader 副本交互。你可以理解为其他副本只是 leader 副本的拷贝，它们的存在只是为了保证消息存储的安全性。

试想一种情况：假如 leader 副本所在的 broker 突然挂掉，那么就要从 follower 副本重新选出一个 leader ，但是 leader 的数据还有一些没有被 follower 副本的同步的话，就会造成消息丢失。

1-设置 acks = all 副本全部收到消息再回调

解决办法就是我们设置 acks = all。acks 是 Kafka 生产者(Producer) 很重要的一个参数。

acks 的默认值即为 1，代表我们的消息被 leader 副本接收之后就算被成功发送。当我们配置 acks = all 表示只有所有 ISR 列表的副本全部收到消息时，生产者才会接收到来自服务器的响应. 这种模式是最高级别的，也是最安全的，可以确保不止一个 Broker 接收到了消息. 该模式的延迟会很高.

2-设置 replication.factor >= 3 副本数

为了保证 leader 副本能有 follower 副本能同步消息，我们一般会为 topic 设置 replication.factor >= 3。这样就可以保证每个 分区(partition) 至少有 3 个副本。虽然造成了数据冗余，但是带来了数据的安全性。

3-设置 min.insync.replicas > 1

一般情况下我们还需要设置 min.insync.replicas> 1 ，这样配置代表消息至少要被写入到 2 个副本才算是被成功发送。min.insync.replicas 的默认值为 1 ，在实际生产中应尽量避免默认值 1。

但是，为了保证整个 Kafka 服务的高可用性，你需要确保 replication.factor > min.insync.replicas 。为什么呢？设想一下假如两者相等的话，只要是有一个副本挂掉，整个分区就无法正常工作了。这明显违反高可用性！一般推荐设置成 replication.factor = min.insync.replicas + 1。

4-设置 unclean.leader.election.enable = false

我们发送的消息会被发送到 leader 副本，然后 follower 副本才能从 leader 副本中拉取消息进行同步。多个 follower 副本之间的消息同步情况不一样，当我们配置了 unclean.leader.election.enable = false 的话，当 leader 副本发生故障时就不会从 follower 副本中和 leader 同步程度达不到要求的副本中选择出 leader ，这样降低了消息丢失的可能性。

【3】保证消息不重复消费

kafka 出现消息重复消费的原因：
（1）服务端侧已经消费的数据没有成功提交 offset（根本原因，消费者没来得及提交offset就挂了）。
（2）Kafka 侧 由于服务端处理业务时间长或者网络链接等等原因让 Kafka 认为服务假死，触发了分区 rebalance。

解决方案：
（1）消费消息服务做幂等校验，比如 Redis 的 set、MySQL 的主键等天然的幂等功能。这种方法最有效。
（2）将 enable.auto.commit 参数设置为 false，关闭自动提交，开发者在代码中手动提交 offset。
那么这里会有个问题：什么时候提交 offset 合适？
1-处理完消息再提交：依旧有消息重复消费的风险，和自动提交一样
2-拉取到消息即提交：会有消息丢失的风险。允许消息延时的场景，一般会采用这种方式。然后，通过定时任务在业务不繁忙（比如凌晨）的时候做数据兜底。

（1）关闭自动提交，开启手动提交offset

spring:kafka:bootstrap-servers: localhost:9092consumer:enable-auto-commit: false  # 关闭自动提交auto-offset-reset: latest  # 可选：设置为latest避免启动时重复消费max-poll-records: 50       # 每次拉取的最大记录数max-poll-interval-ms: 300000  # 处理消息的最大间隔key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializerlistener:ack-mode: manual # 配置手动提交

（2）消费者处理完消息后再手动提交

@Service
public class OrderConsumerService {@Autowiredprivate OrderService orderService;@KafkaListener(topics = "orders-topic", groupId = "order-consumer-group")public void listen(ConsumerRecord<String, String> record, Acknowledgment ack) {try {String orderJson = record.value();OrderDTO orderDTO = JSON.parseObject(orderJson, OrderDTO.class);// 1. 处理订单（幂等性）orderService.processOrder(orderDTO);// 2. 手动提交offsetack.acknowledge();} catch (Exception e) {log.error("处理订单消息失败: {}", e.getMessage(), e);// 根据异常类型决定是否重试或跳过if (e instanceof RetryableException) {// 重试异常，不提交offsetthrow e;} else {// 非重试异常，记录日志并提交offsetlog.error("非重试异常，跳过消息: {}", record.value());ack.acknowledge();}}}
}

（3）添加数据库唯一约束实现幂等消费

消费消息的时候，根据数据库的唯一键判断消息是否已经消费过了，如果已消费就跳过

@Service
public class OrderService {@Autowiredprivate OrderRepository orderRepository;@Autowiredprivate MessageDeduplicationService deduplicationService;@Transactionalpublic void processOrder(OrderDTO orderDTO) {String messageId = orderDTO.getOrderId();// 1. 检查消息是否已处理（Redis去重）if (!deduplicationService.isMessageProcessed(messageId)) {log.info("订单已处理，跳过: {}", messageId);return;}// 2. 再次检查数据库中是否存在（双重检查）Optional<Order> existingOrder = orderRepository.findByOrderId(messageId);if (existingOrder.isPresent()) {log.info("订单已存在，跳过: {}", messageId);return;}// 3. 处理新订单Order order = new Order();order.setOrderId(messageId);order.setAmount(orderDTO.getAmount());order.setStatus("CREATED");// 4. 保存订单orderRepository.save(order);// 5. 后续业务逻辑...processPayment(order);updateInventory(order);}// 其他方法...
}

使用redis校验的逻辑

@Service
public class MessageDeduplicationService {@Autowiredprivate RedisTemplate<String, String> redisTemplate;private static final String DEDUPLICATION_KEY = "processed_messages:";private static final long EXPIRE_TIME = 24 * 60 * 60; // 24小时过期/*** 检查消息是否已处理*/public boolean isMessageProcessed(String messageId) {String key = DEDUPLICATION_KEY + messageId;return Boolean.TRUE.equals(redisTemplate.opsForValue().setIfAbsent(key, "1", EXPIRE_TIME, TimeUnit.SECONDS));}
}

【4】Kafka重试机制

（1）消费失败会怎么样？

在消费过程中，当其中一个消息消费异常时，会不会卡住后续队列消息的消费？这样业务岂不是卡住了？

生产者代码：

 for (int i = 0; i < 10; i++) {kafkaTemplate.send(KafkaConst.TEST_TOPIC, String.valueOf(i))}

消费者消代码：

   @KafkaListener(topics = {KafkaConst.TEST_TOPIC},groupId = "apple")private void customer(String message) throws InterruptedException {log.info("kafka customer:{}",message);Integer n = Integer.parseInt(message);if (n%5==0){throw new  RuntimeException();}}

在默认配置下，当消费异常会进行重试，重试多次后会跳过当前消息，继续进行后续消息的消费，不会一直卡在当前消息。下面是一段消费的日志，可以看出当 test-0@95 重试多次后会被跳过。

2023-08-10 12:03:32.918 DEBUG 9700 --- [ntainer#0-0-C-1] o.s.kafka.listener.DefaultErrorHandler   : Skipping seek of: test-0@95
2023-08-10 12:03:32.918 TRACE 9700 --- [ntainer#0-0-C-1] o.s.kafka.listener.DefaultErrorHandler   : Seeking: test-0 to: 96
2023-08-10 12:03:32.918  INFO 9700 --- [ntainer#0-0-C-1] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=consumer-apple-1, groupId=apple] Seeking to offset 96 for partition test-0

因此，即使某个消息消费异常，Kafka 消费者仍然能够继续消费后续的消息，不会一直卡在当前消息，保证了业务的正常进行。

（2）默认会重试多少次？

默认配置下，消费异常会进行重试，重试次数是多少, 重试是否有时间间隔？

看源码 FailedRecordTracker 类有个 recovered 函数，返回 Boolean 值判断是否要进行重试，下面是这个函数中判断是否重试的逻辑：

	@Overridepublic boolean recovered(ConsumerRecord << ? , ? > record, Exception exception,@Nullable MessageListenerContainer container,@Nullable Consumer << ? , ? > consumer) throws InterruptedException {if (this.noRetries) {// 不支持重试attemptRecovery(record, exception, null, consumer);return true;}// 取已经失败的消费记录集合Map < TopicPartition, FailedRecord > map = this.failures.get();if (map == null) {this.failures.set(new HashMap < > ());map = this.failures.get();}//  获取消费记录所在的Topic和PartitionTopicPartition topicPartition = new TopicPartition(record.topic(), record.partition());FailedRecord failedRecord = getFailedRecordInstance(record, exception, map, topicPartition);// 通知注册的重试监听器，消息投递失败this.retryListeners.forEach(rl - >rl.failedDelivery(record, exception, failedRecord.getDeliveryAttempts().get()));// 获取下一次重试的时间间隔long nextBackOff = failedRecord.getBackOffExecution().nextBackOff();if (nextBackOff != BackOffExecution.STOP) {this.backOffHandler.onNextBackOff(container, exception, nextBackOff);return false;} else {attemptRecovery(record, exception, topicPartition, consumer);map.remove(topicPartition);if (map.isEmpty()) {this.failures.remove();}return true;}}

其中， BackOffExecution.STOP 的值为 -1。

@FunctionalInterface
public interface BackOffExecution {long STOP = -1;long nextBackOff();}

nextBackOff 的值调用 BackOff 类的 nextBackOff() 函数。如果当前执行次数大于最大执行次数则返回 STOP，既超过这个最大执行次数后才会停止重试。

public long nextBackOff() {this.currentAttempts++;if (this.currentAttempts <= getMaxAttempts()) {return getInterval();}else {return STOP;}
}

那么这个 getMaxAttempts 的值又是多少呢？回到最开始，当执行出错会进入 DefaultErrorHandler 。DefaultErrorHandler 默认的构造函数是：

public DefaultErrorHandler() {this(null, SeekUtils.DEFAULT_BACK_OFF);
}

SeekUtils.DEFAULT_BACK_OFF 定义的是:

public static final int DEFAULT_MAX_FAILURES = 10;public static final FixedBackOff DEFAULT_BACK_OFF = new FixedBackOff(0, DEFAULT_MAX_FAILURES - 1);

DEFAULT_MAX_FAILURES 的值是 10，currentAttempts 从 0 到 9，所以总共会执行 10 次，每次重试的时间间隔为 0。

最后，简单总结一下：Kafka 消费者在默认配置下会进行最多 10 次 的重试，每次重试的时间间隔为 0，即立即进行重试。如果在 10 次重试后仍然无法成功消费消息，则不再进行重试，消息将被视为消费失败。

（3）如何自定义重试次数以及时间间

从上面的代码可以知道，默认错误处理器的重试次数以及时间间隔是由 FixedBackOff 控制的，FixedBackOff 是 DefaultErrorHandler 初始化时默认的。所以自定义重试次数以及时间间隔，只需要在 DefaultErrorHandler 初始化的时候传入自定义的 FixedBackOff 即可。重新实现一个 KafkaListenerContainerFactory ，调用 setCommonErrorHandler 设置新的自定义的错误处理器就可以实现。

@Bean
public KafkaListenerContainerFactory kafkaListenerContainerFactory(ConsumerFactory<String, String> consumerFactory) {ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();// 自定义重试时间间隔以及次数FixedBackOff fixedBackOff = new FixedBackOff(1000, 5);factory.setCommonErrorHandler(new DefaultErrorHandler(fixedBackOff));factory.setConsumerFactory(consumerFactory);return factory;
}

（4）如何在重试失败后进行告警?

自定义重试失败后逻辑，需要手动实现，以下是一个简单的例子，重写 DefaultErrorHandler 的 handleRemaining 函数，加上自定义的告警等操作。

@Slf4j
public class DelErrorHandler extends DefaultErrorHandler {public DelErrorHandler(FixedBackOff backOff) {super(null,backOff);}@Overridepublic void handleRemaining(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer, MessageListenerContainer container) {super.handleRemaining(thrownException, records, consumer, container);log.info("重试多次失败");// 自定义操作}
}

DefaultErrorHandler 只是默认的一个错误处理器，Spring Kafka 还提供了 CommonErrorHandler 接口。手动实现 CommonErrorHandler 就可以实现更多的自定义操作，有很高的灵活性。例如根据不同的错误类型，实现不同的重试逻辑以及业务逻辑等

（5）重试失败后的数据如何再次处理?

当达到最大重试次数后，数据会直接被跳过，继续向后进行。当代码修复后，如何重新消费这些重试失败的数据呢？

死信队列（Dead Letter Queue，简称 DLQ） 是消息中间件中的一种特殊队列。它主要用于处理无法被消费者正确处理的消息，通常是因为消息格式错误、处理失败、消费超时等情况导致的消息被"丢弃"或"死亡"的情况。当消息进入队列后，消费者会尝试处理它。如果处理失败，或者超过一定的重试次数仍无法被成功处理，消息可以发送到死信队列中，而不是被永久性地丢弃。在死信队列中，可以进一步分析、处理这些无法正常消费的消息，以便定位问题、修复错误，并采取适当的措施。

@RetryableTopic 是 Spring Kafka 中的一个注解,它用于配置某个 Topic 支持消息重试，更推荐使用这个注解来完成重试。

// 重试 5 次，重试间隔 100 毫秒,最大间隔 1 秒
@RetryableTopic(attempts = "5",backoff = @Backoff(delay = 100, maxDelay = 1000)
)
@KafkaListener(topics = {KafkaConst.TEST_TOPIC}, groupId = "apple")
private void customer(String message) {log.info("kafka customer:{}", message);Integer n = Integer.parseInt(message);if (n % 5 == 0) {throw new RuntimeException();}System.out.println(n);
}

当达到最大重试次数后，如果仍然无法成功处理消息，消息会被发送到对应的死信队列中。对于死信队列的处理，既可以用 @DltHandler 处理，也可以使用 @KafkaListener 重新消费。