本文介绍通过合理配置 concurrentmessagelistenercontainer 的并发数与分区绑定关系，实现对单一分区的独立 offset 提交，避免跨分区干扰，确保精准、可控的消费确认语义。
本文介绍通过合理配置 concurrentmessagelistenercontainer 的并发数与分区绑定关系，实现对单一分区的独立 offset 提交，避免跨分区干扰，确保精准、可控的消费确认语义。
在 Spring Kafka 中，Acknowledgement.acknowledge() 默认会对当前线程所处理的整个拉取批次（batch）中所有记录所属的分区统一提交 offset —— 但它本身不支持指定 TopicPartition。因此，若一个消费者实例同时监听多个分区（如 partitions = {"0", "1"}），默认行为会导致“一荣俱荣、一损俱损”：即使仅 partition-0 处理成功，调用 acknowledge() 也会将 partition-1 的 offset 一并提交（前提是该批次中包含其记录），从而引发重复消费或数据丢失风险。
根本解法在于：让每个分区由独立的 KafkaMessageListenerContainer 实例专属处理。Spring Kafka 的 ConcurrentMessageListenerContainer 在显式指定 @TopicPartition 时，会根据 concurrency 值自动将分区均匀分配给子容器。关键规则如下：
- 若 topicPartitions 显式声明了 N 个分区（例如 {"0", "1", "2"}），且 concurrency >= N，则框架会为每个分区创建一个专属的子容器；
- 每个子容器内部只消费单一固定分区的数据，因此 @KafkaListener 方法接收到的 records 列表必然全部来自同一 TopicPartition；
- 此时调用 acknowledgement.acknowledge() 将仅提交该分区的 offset，完全符合预期。
✅ 正确配置示例：
@Configuration
@EnableKafka
public class KafkaConfig {
@Bean
KafkaListenerContainerFactory<concurrentmessagelistenercontainer string>>
kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<integer string> factory =
new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerFactory());
// ⚠️ 关键：concurrency 必须 ≥ 显式声明的分区总数
factory.setConcurrency(3); // 对应 topic1-partitions: [0,1] + topic2-partition: [0,1] → 共4个分区？需设为4！
factory.getContainerProperties().setPollTimeout(3000);
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL); // 显式启用手动确认
return factory;
}
// ... consumerFactory(), consumerConfigs() 等保持不变
}</integer></concurrentmessagelistenercontainer>
✅ 对应的监听器（按分区粒度拆分）：
// 每个 @KafkaListener 绑定唯一分区 → 每个子容器只处理一个分区
@KafkaListener(
id = "topic1-partition0",
topicPartitions = @TopicPartition(topic = "topic1", partitions = "0")
)
public void listenTopic1Partition0(List<consumerrecord string>> records,
Acknowledgement ack) {
try {
records.forEach(record -> process(record)); // 业务处理
ack.acknowledge(); // ✅ 安全：仅提交 topic1-0 的 offset
} catch (Exception e) {
// 记录错误，不 acknowledge → 该分区将重试
log.error("Failed to process partition 0", e);
}
}
@KafkaListener(
id = "topic1-partition1",
topicPartitions = @TopicPartition(topic = "topic1", partitions = "1")
)
public void listenTopic1Partition1(List<consumerrecord string>> records,
Acknowledgement ack) {
// 同理，仅影响 topic1-1
processBatch(records);
ack.acknowledge();
}</consumerrecord></consumerrecord>
? 注意事项：
- concurrency 必须 ≥ 显式声明的分区总数，否则 Spring Kafka 会自动降级并发数，并复用容器处理多个分区，导致 acknowledge() 仍影响多分区；
- 推荐配合 AckMode.MANUAL 使用（如上例所示），避免 BATCH 或 RECORD 模式下隐式提交带来的歧义；
- 若使用动态分区分配（即不写死 @TopicPartition），则无法实现分区级精确控制，需改用 ConsumerAwareRebalanceListener + 手动 consumer.commitSync(Map)；
- 日志中留意警告："When specific partitions are provided, the concurrency must be less than or equal to the number of partitions..." —— 这说明配置已生效。
总结：Spring Kafka 的分区级 Offset 控制并非依赖 Acknowledgement 的扩展参数，而是通过架构层隔离（每个分区独占一个 listener container） 实现的。合理设置 concurrency 并显式声明分区，即可让 acknowledge() 天然具备分区粒度语义，兼顾简洁性与可靠性。
