当前位置：首页 > web >正文

Kafka 面试题及详细答案100道（23-35）-- 核心机制2

web 2025/8/18 9:26:39

《前后端面试题》专栏集合了前后端各个知识模块的面试题，包括html，javascript，css，vue，react，java，Openlayers，leaflet，cesium，mapboxGL，threejs，nodejs，mangoDB，SQL，Linux… 。

前后端面试题-专栏总目录

在这里插入图片描述

文章目录

一、本文面试题目录
- - 23. 什么是消息的偏移量（Offset）？它有什么作用？
  - 24. 消费者如何提交偏移量（Offset Commit）？有哪些提交方式？
  - 25. 自动提交偏移量和手动提交偏移量各有什么优缺点？
  - 26. 什么是Kafka的日志分段（Log Segments）？它的作用是什么？
  - 27. Kafka的日志清理策略有哪些？分别适用于什么场景？
  - 28. 什么是Kafka的压缩机制？支持哪些压缩算法？
  - 29. 生产者如何选择将消息发送到哪个分区？有哪些分区策略？
  - 30. 简述Kafka的幂等性（Idempotence）机制，如何保证消息不重复？
  - 31. 什么是事务（Transactions）在Kafka中的应用？如何实现？
  - 32. Kafka的消息投递语义有哪些（至少一次、至多一次、 exactly once）？如何保证这些语义？
  - 33. 什么是Kafka的控制器（Controller）？它的作用是什么？
  - 34. 控制器是如何选举产生的？
  - 35. 简述Kafka的延时操作机制（如延时队列的实现原理）。
二、100道Kafka 面试题目录列表

一、本文面试题目录

23. 什么是消息的偏移量（Offset）？它有什么作用？

消息的偏移量（Offset） 是Kafka中标识消息在分区内位置的唯一整数，从0开始递增。每条消息被追加到分区时，都会被分配一个唯一的offset，且永远不会重复。

偏移量的主要作用：

唯一标识消息：在一个分区内，offset是消息的唯一标识符
记录消费位置：消费者通过记录offset跟踪自己的消费进度
保证消息顺序：offset的递增性保证了分区内消息的顺序性
支持消息回溯：消费者可以通过指定offset重新消费历史消息
日志分段依据：Kafka根据offset范围将日志分为多个Segment

示例：消费者指定offset消费消息

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.util.*;public class OffsetExample {public static void main(String[] args) {Properties props = new Properties();props.put("bootstrap.servers", "localhost:9092");props.put("group.id", "offset-example-group");props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);String topic = "user-actions";// 订阅主题consumer.subscribe(Arrays.asList(topic));// 手动指定从offset=100开始消费Set<TopicPartition> partitions = consumer.assignment();// 等待分配分区while (partitions.isEmpty()) {consumer.poll(Duration.ofMillis(100));partitions = consumer.assignment();}// 为每个分区设置起始offsetfor (TopicPartition partition : partitions) {consumer.seek(partition, 100);}// 消费消息while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : records) {System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());}}}
}

24. 消费者如何提交偏移量（Offset Commit）？有哪些提交方式？

消费者提交偏移量（Offset Commit）是指将已消费消息的offset记录到Kafka（__consumer_offsets主题）的过程，用于跟踪消费进度。

主要提交方式：

自动提交：
- 由消费者定期自动提交最新的offset
- 通过enable.auto.commit=true启用
- 提交间隔由auto.commit.interval.ms控制（默认5秒）
手动同步提交：
- 调用consumer.commitSync()方法显式提交
- 会阻塞当前线程，直到提交成功或抛出异常
- 适用于需要确保提交成功的场景
手动异步提交：
- 调用consumer.commitAsync()方法异步提交
- 不会阻塞线程，通过回调函数处理结果
- 适用于对吞吐量要求高，可容忍偶尔提交失败的场景
指定偏移量提交：
- 手动指定要提交的offset值
- 适用于精确控制提交位置的场景

示例：不同的偏移量提交方式

// 1. 自动提交配置
Properties autoCommitProps = new Properties();
autoCommitProps.put("enable.auto.commit", "true");
autoCommitProps.put("auto.commit.interval.ms", "5000"); // 每5秒自动提交// 2. 手动同步提交
void syncCommitExample(KafkaConsumer<String, String> consumer) {try {while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));// 处理消息for (ConsumerRecord<String, String> record : records) {processRecord(record);}// 同步提交consumer.commitSync();}} finally {consumer.close();}
}// 3. 手动异步提交
void asyncCommitExample(KafkaConsumer<String, String> consumer) {try {while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));// 处理消息for (ConsumerRecord<String, String> record : records) {processRecord(record);}// 异步提交并处理结果consumer.commitAsync((offsets, exception) -> {if (exception != null) {System.err.println("Commit failed for offsets: " + offsets);exception.printStackTrace();}});}} finally {consumer.close();}
}// 4. 指定偏移量提交
void specificOffsetCommitExample(KafkaConsumer<String, String> consumer) {Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : records) {processRecord(record);// 记录每个分区的最新offsetoffsets.put(new TopicPartition(record.topic(), record.partition()),new OffsetAndMetadata(record.offset() + 1)); // +1表示下一个要消费的位置}// 提交指定的offsetconsumer.commitSync(offsets);}
}

25. 自动提交偏移量和手动提交偏移量各有什么优缺点？

自动提交偏移量

优点：

配置简单，无需编写额外代码
适合简单场景，减少开发复杂度
不需要关心提交时机和逻辑

缺点：

可能导致重复消费：如果在自动提交前消费者崩溃，重启后会重新消费已处理但未提交的消息
可能导致消息丢失：如果在处理消息前自动提交了offset，处理失败后不会重新消费
灵活性低，无法根据业务逻辑精确控制提交时机
不适合事务性场景或需要精确处理的场景

手动提交偏移量

优点：

可以精确控制提交时机，通常在消息处理完成后提交
减少重复消费或消息丢失的风险
支持更复杂的消费逻辑，如批量处理后提交
适合事务性场景和对消息处理准确性要求高的场景

缺点：

需要编写更多代码，增加开发复杂度
需要处理提交失败的情况
可能因提交不及时导致重平衡时重复消费增多
需要合理设计提交频率，平衡性能和可靠性

适用场景建议：

自动提交：简单的日志收集、非关键业务数据处理
手动提交：金融交易、订单处理等关键业务，需要确保消息处理的准确性

26. 什么是Kafka的日志分段（Log Segments）？它的作用是什么？

Kafka的日志分段（Log Segments） 是指将每个分区的日志（Log）分割成多个大小相等的文件片段进行存储。每个分区的日志目录下包含多个Segment文件，每个Segment由一个数据文件（.log）和两个索引文件（.index和.timeindex）组成。

日志分段的主要作用：

便于日志管理：
- 避免单个文件过大，提高文件操作效率
- 便于实现日志的保留和清理策略
提高查找效率：
- 通过索引文件快速定位消息位置
- 减少单个文件的IO操作开销
支持并行操作：
- 不同Segment可以被并行读取和写入
- 清理旧Segment时不影响新消息的写入
优化存储利用：
- 可以针对不同时间段的Segment采用不同的存储策略
- 便于实现数据压缩和归档

Segment的工作机制：

每个分区有一个活跃Segment（Active Segment），负责接收新消息
当活跃Segment达到配置的大小（log.segment.bytes，默认1GB）或时间（log.roll.hours）时，会创建新的Segment
旧的Segment会根据保留策略被删除或压缩
索引文件记录了消息offset与物理文件位置的映射关系

示例：配置日志分段相关参数（server.properties）

# 每个Segment文件的最大大小，默认1GB
log.segment.bytes=1073741824# 强制滚动生成新Segment的时间，默认7天
log.roll.hours=168# 索引文件中每条记录的大小，默认4字节
log.index.size.max.bytes=10485760# 索引项之间的最大偏移量差距
log.index.interval.bytes=4096

27. Kafka的日志清理策略有哪些？分别适用于什么场景？

Kafka提供了多种日志清理策略，用于管理磁盘空间并删除不再需要的数据：

删除策略（delete，默认）：
- 原理：当日志达到指定的保留时间或大小后，删除整个旧的Segment文件
- 配置参数：
  - log.retention.hours：按时间保留（默认168小时/7天）
  - log.retention.bytes：按总大小保留（默认-1，无限制）
  - log.retention.check.interval.ms：检查间隔
- 适用场景：
  - 日志收集、事件跟踪等只需要保留一段时间的场景
  - 消息是时序性的，且不需要按Key保留最新值的场景
压缩策略（compact）：
- 原理：保留每个Key的最新版本消息，删除相同Key的旧消息
- 配置参数：
  - log.cleaner.enable=true：启用清理器
  - log.cleanup.policy=compact：设置为压缩策略
  - log.cleaner.min.compaction.lag.ms：最小压缩延迟
- 适用场景：
  - 存储需要更新的数据，如用户配置、产品信息
  - 变更日志（Change Log）场景，需要保留最新状态
  - 键值存储的备份或缓存
删除+压缩混合策略：
- 原理：结合了删除和压缩两种策略
- 配置：log.cleanup.policy=delete,compact
- 适用场景：需要保留最新状态但也有时间限制的场景

示例：为不同Topic配置清理策略

# 创建一个使用压缩策略的Topic
bin/kafka-topics.sh --create \--bootstrap-server localhost:9092 \--topic user-profiles \--partitions 3 \--replication-factor 2 \--config cleanup.policy=compact \--config retention.ms=604800000  # 同时保留7天# 修改已有Topic的清理策略
bin/kafka-configs.sh --alter \--bootstrap-server localhost:9092 \--topic system-logs \--add-config cleanup.policy=delete,retention.hours=24

28. 什么是Kafka的压缩机制？支持哪些压缩算法？

Kafka的压缩机制是指在生产者端对消息进行压缩后再发送，以减少网络传输量和存储开销的技术。压缩和解压缩操作主要在生产者和消费者端进行，Broker仅负责存储压缩后的消息。

压缩机制的工作原理：

生产者将多条消息批量压缩成一个压缩块
压缩后的消息被发送到Broker并存储
Broker不解压消息，直接按压缩格式存储
消费者从Broker获取压缩消息后进行解压并处理

支持的压缩算法：

GZIP：
- 压缩率高，适合CPU资源充足但带宽有限的场景
- 压缩和解压速度相对较慢
Snappy：
- 压缩率适中，速度快，CPU占用低
- 由Google开发，适合大多数场景，是推荐的默认选择
LZ4：
- 压缩速度非常快，压缩率比Snappy稍低
- 适合对性能要求极高的场景
ZSTD（Kafka 2.1.0+支持）：
- 压缩率优于GZIP，速度接近Snappy
- 提供更好的压缩比和性能平衡

示例：配置生产者压缩算法

// Java生产者配置压缩算法
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");// 设置压缩算法
props.put("compression.type", "snappy"); // 可选值: none, gzip, snappy, lz4, zstd// 配置批量处理参数以提高压缩效率
props.put("linger.ms", 5); // 等待5ms再发送，积累更多消息
props.put("batch.size", 16384); // 批量大小，默认16KBKafkaProducer<String, String> producer = new KafkaProducer<>(props);

# Python生产者配置压缩算法
from kafka import KafkaProducer
import jsonproducer = KafkaProducer(bootstrap_servers=['localhost:9092'],value_serializer=lambda v: json.dumps(v).encode('utf-8'),compression_type='snappy',  # 设置压缩算法linger_ms=5,batch_size=16384
)

29. 生产者如何选择将消息发送到哪个分区？有哪些分区策略？

Kafka生产者发送消息时，需要确定消息被分配到Topic的哪个Partition。分区策略决定了消息的分布方式，直接影响负载均衡和消息顺序性。

分区选择流程：

如果消息指定了分区号（partition），则直接发送到该分区
如果未指定分区但指定了Key，则通过Key的哈希值确定分区
如果既未指定分区也未指定Key，则采用轮询（Round Robin）方式分配分区

主要分区策略：

默认分区策略：
- 由DefaultPartitioner实现
- 对Key进行哈希计算（murmur2算法），然后与分区数取模
- 相同Key的消息会被分配到同一分区，保证顺序性
轮询分区策略：
- 不考虑Key，按顺序将消息分配到不同分区
- 实现负载均衡，但无法保证相同Key的消息在同一分区
自定义分区策略：
- 实现Partitioner接口，根据业务需求定制分区逻辑
- 例如：按地理位置、用户ID范围、业务类型等分区

示例：使用不同的分区策略

// 1. 默认分区策略（按Key分区）
ProducerRecord<String, String> recordWithKey = new ProducerRecord<>("user-tracking", "user123", "click-event");
// 相同Key会被分到同一个分区
producer.send(recordWithKey);// 2. 无Key的轮询策略
ProducerRecord<String, String> recordWithoutKey = new ProducerRecord<>("system-logs", null, "error-message");
// 会被轮询分配到不同分区
producer.send(recordWithoutKey);// 3. 自定义分区策略
public class RegionPartitioner implements Partitioner {@Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {// 获取分区列表List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);int numPartitions = partitions.size();// 假设Key是地区代码if (keyBytes == null) {throw new InvalidRecordException("Key (region) must not be null");}String region = (String) key;// 按地区分配到不同分区switch (region) {case "north":return 0 % numPartitions;case "south":return 1 % numPartitions;case "east":return 2 % numPartitions;case "west":return 3 % numPartitions;default:return Math.abs(Utils.murmur2(keyBytes)) % numPartitions;}}@Overridepublic void close() {}@Overridepublic void configure(Map<String, ?> configs) {}
}// 使用自定义分区器
Properties props = new Properties();
// 其他配置...
props.put("partitioner.class", "com.example.RegionPartitioner");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);

30. 简述Kafka的幂等性（Idempotence）机制，如何保证消息不重复？

Kafka的幂等性机制是指生产者能够保证多次发送相同的消息，Broker只会持久化一次，从而避免消息重复。这对于确保数据一致性至关重要，尤其是在网络不稳定或系统故障的情况下。

幂等性实现原理：

生产者ID（PID）：
- 每个生产者实例启动时会被分配一个唯一的PID
- 重启生产者会生成新的PID
序列号（Sequence Number）：
- 生产者为发送到每个分区的消息分配一个递增的序列号
- 每个分区的序列号独立计数，从0开始
Broker端去重：
- Broker为每个<PID, 分区>对维护一个序列号
- 当收到消息时，检查消息的序列号：
  - 如果大于当前值：接受消息并更新序列号
  - 如果等于当前值：说明是重试消息，忽略但返回成功
  - 如果小于当前值：说明是过期消息，直接拒绝

启用幂等性：
只需在生产者配置中设置enable.idempotence=true，Kafka会自动处理其余细节。

示例：配置幂等性生产者

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");// 启用幂等性
props.put("enable.idempotence", "true");// 幂等性需要设置至少1的acks
props.put("acks", "all");// 重试次数，建议设置较大值
props.put("retries", 3);// 重试间隔
props.put("retry.backoff.ms", 100);KafkaProducer<String, String> producer = new KafkaProducer<>(props);

注意事项：

幂等性仅保证单个生产者对单个分区的Exactly-Once语义
跨分区或跨生产者的消息重复需要通过事务机制解决
生产者重启后会生成新的PID，因此幂等性不能跨会话保证

31. 什么是事务（Transactions）在Kafka中的应用？如何实现？

Kafka中的事务（Transactions） 机制允许生产者原子性地发送一批消息到一个或多个Topic，同时支持消费者原子性地提交消费偏移量，确保消息处理的一致性。

事务的主要应用场景：

原子性地发送多条消息到多个Topic或分区
原子性地消费消息并生产处理结果（如流处理中的Exactly-Once语义）
确保跨多个操作的数据一致性

事务实现原理：

事务协调器（Transaction Coordinator）：
- 每个生产者通过PID关联到一个事务协调器
- 负责管理事务状态和提交/中止事务
事务日志（Transaction Log）：
- 一个特殊的内部Topic（__transaction_state）
- 存储所有事务的状态信息
事务ID（Transaction ID）：
- 应用程序指定的唯一ID，用于跨生产者会话标识事务
- 确保生产者重启后仍能恢复事务状态
状态标记：
- 事务中的消息会被标记为"prepared"状态
- 事务提交后标记为"committed"，消费者可见
- 事务中止后标记为"aborted"，消费者会忽略

示例：使用Kafka事务

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");// 启用事务需要的配置
props.put("enable.idempotence", "true");
props.put("transactional.id", "order-processing-transaction-1"); // 事务IDKafkaProducer<String, String> producer = new KafkaProducer<>(props);// 初始化事务
producer.initTransactions();try {// 开始事务producer.beginTransaction();// 发送一批消息producer.send(new ProducerRecord<>("orders", "order1", "order-details-1"));producer.send(new ProducerRecord<>("inventory-updates", "item1", "decrement-1"));// 如果需要，还可以提交消费偏移量// producer.sendOffsetsToTransaction(offsets, "consumer-group-id");// 提交事务producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {// 这些异常无法恢复，需要关闭生产者producer.close();
} catch (KafkaException e) {// 中止事务producer.abortTransaction();
}producer.close();

消费者需要配置隔离级别以正确处理事务消息：

Properties consumerProps = new Properties();
// 其他配置...// 设置事务隔离级别
// read_committed: 只消费已提交的消息（默认）
// read_uncommitted: 消费所有消息，包括未提交的
consumerProps.put("isolation.level", "read_committed");

32. Kafka的消息投递语义有哪些（至少一次、至多一次、 exactly once）？如何保证这些语义？

Kafka提供三种消息投递语义，描述消息从生产者到消费者的传递保证：

至多一次（At Most Once）：
- 定义：消息可能丢失，但不会被重复处理
- 实现方式：
  - 生产者：不重试失败的发送
  - 消费者：自动提交偏移量，且提交时机早于消息处理
- 优点：性能好，无重复
- 缺点：可能丢失消息
- 适用场景：非关键数据，如监控指标、日志收集
至少一次（At Least Once）：
- 定义：消息不会丢失，但可能被重复处理
- 实现方式：
  - 生产者：启用重试（retries > 0）和acks=all
  - 消费者：手动提交偏移量，在消息处理完成后提交
- 优点：消息不丢失
- 缺点：可能重复处理，需要业务层实现幂等性
- 适用场景：大多数业务场景，如订单处理、支付通知
精确一次（Exactly Once）：
- 定义：消息恰好被处理一次，既不丢失也不重复
- 实现方式：
  - 生产者：启用幂等性（enable.idempotence=true）和事务（transactional.id）
  - 消费者：结合事务机制提交偏移量
  - 流处理：使用Kafka Streams的Exactly-Once语义
- 优点：数据一致性最高
- 缺点：性能开销大，实现复杂
- 适用场景：金融交易、关键业务数据等对一致性要求极高的场景

示例：配置不同的投递语义

// 1. 配置至多一次语义
Properties atMostOnceProps = new Properties();
// 生产者配置
atMostOnceProps.put("retries", 0); // 不重试
// 消费者配置
atMostOnceProps.put("enable.auto.commit", "true");
atMostOnceProps.put("auto.commit.interval.ms", "1000");// 2. 配置至少一次语义
Properties atLeastOnceProps = new Properties();
// 生产者配置
atLeastOnceProps.put("retries", 3);
atLeastOnceProps.put("acks", "all");
// 消费者配置
atLeastOnceProps.put("enable.auto.commit", "false"); // 手动提交// 3. 配置精确一次语义
Properties exactlyOnceProps = new Properties();
// 生产者配置
exactlyOnceProps.put("enable.idempotence", "true");
exactlyOnceProps.put("transactional.id", "my-transaction-id");
exactlyOnceProps.put("acks", "all");
exactlyOnceProps.put("retries", 3);
// 消费者配置
exactlyOnceProps.put("isolation.level", "read_committed");

33. 什么是Kafka的控制器（Controller）？它的作用是什么？

Kafka的控制器（Controller） 是Kafka集群中的一个特殊Broker，负责管理整个集群的分区和副本状态。在Kafka集群中，只有一个控制器在工作，其他Broker作为备用。

控制器的主要作用：

分区Leader选举：
- 当创建新分区或Leader副本故障时，负责选举新的Leader
- 确保每个分区始终有一个可用的Leader副本
集群元数据管理：
- 维护集群的元数据信息（Broker、Topic、Partition、副本等）
- 向其他Broker同步元数据变化
Broker故障处理：
- 监控集群中所有Broker的状态
- 当检测到Broker故障时，触发受影响分区的Leader重新选举
分区重分配协调：
- 协调分区在Broker之间的重分配操作
- 确保重分配过程中服务不中断
主题管理操作：
- 处理创建、删除Topic的请求
- 处理增加Topic分区的请求

控制器的工作原理：

控制器通过ZooKeeper的临时节点实现选举和故障检测
控制器与其他Broker保持通信，同步集群状态
控制器决定何时触发Leader选举，并将结果通知所有Broker

34. 控制器是如何选举产生的？

Kafka控制器的选举过程主要依赖ZooKeeper的分布式锁机制，确保集群中始终有且仅有一个控制器在工作：

选举过程：

初始化阶段：
- 集群启动时，所有Broker都尝试在ZooKeeper上创建/controller临时节点
- 第一个成功创建该节点的Broker成为控制器
节点内容：
- /controller节点存储控制器的信息，格式为：
```
{"version":1, "brokerid":1, "timestamp":"1620000000000"}
```
- 包含控制器的Broker ID和当选时间戳
故障检测：
- 所有Broker都监听/controller节点的变化
- 当现有控制器故障时，其与ZooKeeper的会话超时，/controller节点被删除
重新选举：
- /controller节点被删除后，触发所有存活Broker的重新选举
- 每个Broker再次尝试创建/controller节点
- 第一个成功创建节点的Broker成为新的控制器
控制器初始化：
- 新控制器当选后，读取ZooKeeper中的集群元数据
- 向所有Broker发送请求，获取最新的分区状态
- 建立与其他Broker的通信通道，开始履行控制器职责

无ZooKeeper模式（Kafka Raft Metadata Mode）：

Kafka 2.8.0引入了KRaft模式，用Kafka自身的Raft协议替代ZooKeeper
控制器选举由Raft协议管理，更高效且减少了外部依赖
一组专门的Broker（Controller Quorum）负责控制器选举和元数据管理

35. 简述Kafka的延时操作机制（如延时队列的实现原理）。

Kafka本身没有专门的延时队列功能，但可以通过一些机制实现延时操作，主要依赖于其内部的延时请求处理器和时间轮（Timing Wheel）数据结构。

Kafka延时操作的实现原理：

时间轮（Timing Wheel）：
- 一种高效的定时任务管理数据结构，类似时钟的轮盘
- 由多个bucket组成，每个bucket代表一个时间间隔
- 每个bucket中存储该时间段内需要执行的任务
延时请求处理器：
- Kafka内部使用DelayedOperationPurgatory管理所有延时操作
- 支持多种延时请求类型：如生产请求超时、拉取请求超时等
延时队列的实现方式：
- 时间戳过滤法：
  1. 生产者发送消息时，在消息中包含目标消费时间戳
  2. 消费者拉取消息后，检查时间戳是否到达
  3. 未到达则暂存消息，等待时间到达后再处理
- 主题分区法：
  1. 创建多个代表不同延时级别的Topic（如delay-10s, delay-1min等）
  2. 生产者根据需要的延时级别发送到对应的Topic
  3. 专门的消费者消费这些Topic，等待指定时间后转发到目标Topic

示例：基于时间戳的延时队列实现

// 生产者发送带有延时信息的消息
ProducerRecord<String, String> record = new ProducerRecord<>("delayed-messages", "order123", "{\"data\":\"order details\", \"delayUntil\":1620000000000}" // 目标时间戳
);
producer.send(record);// 消费者处理延时消息
while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));long now = System.currentTimeMillis();for (ConsumerRecord<String, String> record : records) {// 解析消息获取延时目标时间JsonObject json = new JsonParser().parse(record.value()).getAsJsonObject();long delayUntil = json.get("delayUntil").getAsLong();if (now >= delayUntil) {// 时间已到，处理消息processMessage(json.get("data").getAsString());// 提交偏移量consumer.commitSync();} else {// 时间未到，暂存消息稍后处理scheduleForLaterProcessing(record, delayUntil - now);// 不提交偏移量，以便下次继续处理}}// 处理已到时间的暂存消息processScheduledMessages();
}

注意事项：

Kafka的延时队列实现并非原生支持，存在一定局限性
时间戳过滤法可能导致消费者重新平衡时重复处理消息
对于精确的延时队列需求，可考虑结合专门的延时队列系统（如RabbitMQ的延时交换机）

二、100道Kafka 面试题目录列表

文章序号	Kafka 100道
1	Kafka面试题及详细答案100道（01-10）
2	Kafka面试题及详细答案100道（11-22）
3	Kafka面试题及详细答案100道（23-35）
4	Kafka面试题及详细答案100道（36-50）
5	Kafka面试题及详细答案100道（51-65）
6	Kafka面试题及详细答案100道（66-80）
7	Kafka面试题及详细答案100道（81-90）
8	Kafka面试题及详细答案100道（91-95）
9	Kafka面试题及详细答案100道（96-100）