Kafka生产者和消费者核心参数调优实战：从原理到高可靠订单系统应用

1. 从一次线上告警说起：为什么你的Kafka消息时快时慢？

那天下午，监控系统突然弹出一条告警：订单处理流水线出现积压，延迟从平时的几十毫秒飙升到了十几秒。团队立刻进入排查状态，数据库连接池、应用服务器负载、下游服务响应时间……一通检查下来，指标都显示正常。最后，我们把目光锁定在了消息队列——Kafka上。通过查看生产者和消费者的监控指标，发现了一个有趣的现象：生产端的发送耗时（request-latency-avg）非常平稳，但消费端的records-lag（消费滞后）却在间歇性跳涨。

问题根源很快浮出水面：生产者在以极快的速度批量推送大消息，而消费者的配置却过于“保守”，一次只拉取少量数据，并且处理逻辑是同步的。这就好比一个火力全开的高压水枪（生产者）在向一个细小的漏斗（消费者）注水，水枪喷得再猛，漏斗的吞吐量上不去，水（消息）自然就会在漏斗口堆积、溢出（延迟）。

这次经历让我深刻意识到，Kafka的高性能，绝不仅仅是搭建一个集群就能自动获得的。它更像一辆顶级跑车，生产者是油门，消费者是变速箱和轮胎，参数配置就是你的驾驶模式。如果油门（生产者）和变速箱（消费者）的配合不当，要么跑不起来，要么容易失控。很多开发者，包括早期的我，往往只满足于“能发能收”，却忽略了参数调优这片深水区，直到线上出问题才追悔莫及。

今天，我们就抛开那些笼统的概念，深入到Java客户端的代码层面，手把手拆解Kafka生产者和消费者的核心参数配置。我会结合真实的踩坑案例，告诉你每个参数背后的设计逻辑、它如何影响性能与可靠性，以及在不同业务场景下该如何权衡和选择。无论你是正在准备面试，还是希望优化现有系统，这篇文章都能给你提供可直接“抄作业”的配置思路和避坑指南。

2. 生产者配置：不只是把消息扔出去那么简单

很多人把Kafka生产者想象成一个简单的“发送者”，配置几个服务器地址和序列化器就完事了。但实际上，现代Kafka生产者是一个高度优化、异步操作的复杂客户端。它的核心目标是在高吞吐量、低延迟和消息可靠性之间找到最佳平衡点。错误的配置，轻则导致性能瓶颈，重则引发数据丢失。

2.1 核心三板斧：acks,linger.ms,batch.size

这三个参数是生产者调优的基石，它们共同决定了消息发送的“节奏”和“保证”。

acks：消息的“安全等级”确认这个参数定义了生产者认为消息“发送成功”的标准。它直接关系到数据的可靠性和吞吐量。

• acks=0：“发了就算”模式。生产者发送消息后，完全不等待任何来自服务器的确认，立即认为发送成功。这是吞吐量最高、延迟最低的模式，但也是可靠性最差的。因为网络闪断、Broker宕机都可能导致消息无声无息地丢失。适用场景：对可靠性要求极低的日志收集、 metrics 上报，丢失少量数据无关紧要。
• acks=1：“Leader确认”模式（默认值）。生产者等待分区的Leader副本将消息写入其本地日志后，就返回成功。这是一个很好的折中方案。它避免了acks=0的完全不可靠，又比acks=all更快。风险在于，如果Leader刚写入就宕机，且该消息还未被其他Follower同步，那么这条消息就会丢失。
• acks=all(或acks=-1)：“全副本确认”模式。生产者需要等待ISR（In-Sync Replicas， 同步副本）列表中的所有副本都成功写入消息后，才返回成功。这是可靠性最高的模式，可以保证只要有一个ISR副本存活，消息就不会丢失。但代价是延迟最高、吞吐量最低。min.insync.replicas（通常设置在Broker端）参数与之配合，定义了最小ISR数量，如果可用ISR数量小于此值，生产者会收到NotEnoughReplicasException异常。

踩坑实录：我们有一个金融对账服务，最初使用acks=1。在一次Broker滚动重启时，某个分区的Leader切换，导致少量处于“已写入Leader但未同步Follower”状态的消息丢失，造成了资金流水对账不平。事后我们将该生产者的acks改为all，并将Broker的min.insync.replicas设为2，从此再未发生类似问题。教训：对数据强一致性的场景，acks=all是必须的。

linger.ms与batch.size：吞吐量的“加速器”Kafka生产者并不是来一条消息就发一条，而是会先放入一个内存缓冲区（RecordAccumulator），等待批量发送。这两个参数就是控制“何时发送这个批次”的。

• linger.ms： 批次等待时间，默认0。即使批次没满，等待这个时间后也会发送。增加此值（例如设为5或10毫秒）可以显著增加批量发送的机会，从而提升吞吐量，但会以增加少量延迟为代价。
• batch.size： 批次大小，默认16KB。当批次中消息的总大小达到此阈值时，会立即发送。增大此值（例如设为64KB或128KB）同样能提升吞吐量，但需要更多内存。

它们是如何协同工作的？生产者会为每个分区维护一个批次。满足以下任一条件，批次就会被发送：

1. 批次大小达到batch.size。
2. 距离上次发送时间超过linger.ms。
3. 缓冲区满了（由buffer.memory控制，默认32MB）。
4. 有新的批次需要当前批次占用的分区（比如分区leader变更）。

Properties props = new Properties(); props.put("bootstrap.servers", "kafka1:9092,kafka2:9092"); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 可靠性优先配置（如订单、交易） props.put("acks", "all"); // 最强可靠性 props.put("max.in.flight.requests.per.connection", 1); // 配合acks=all，保证顺序 props.put("retries", Integer.MAX_VALUE); // 无限重试 props.put("enable.idempotence", true); // 启用幂等性，避免重复 // 吞吐量优先配置（如日志、行为追踪） props.put("acks", "1"); // 平衡可靠性与性能 props.put("linger.ms", 10); // 等待10ms聚合批次 props.put("batch.size", 65536); // 64KB批次 props.put("compression.type", "snappy"); // 启用压缩，减少网络IO // buffer.memory 可根据峰值流量适当调大，如 64MB props.put("buffer.memory", 67108864); KafkaProducer<String, String> producer = new KafkaProducer<>(props);

2.2 高级特性与容错配置

幂等性 (enable.idempotence)与事务

• 幂等性： 设置为true后，生产者会自动将acks设为all，max.in.flight.requests.per.connection设为5（或更小），并启用内部序列号机制。这可以保证单分区、单会话内消息不重复（即“恰好一次”语义的基础）。对于retries可能引起的重复问题，这是一个优雅的解决方案。
• 事务： 用于跨多个分区和消费者组的“原子性”写入。需要配置transactional.id。这对于类似“发布消息同时更新数据库”这种需要跨系统一致性的场景至关重要，但会带来额外的性能开销。

max.in.flight.requests.per.connection：顺序与吞吐的权衡这个参数控制生产者在收到服务器响应之前，最多可以发送多少个未确认的请求，默认值为5。增大它可以提升吞吐（管道更满），但在acks=0或acks=1且启用了重试（retries > 0）时，可能导致分区内的消息顺序错乱。因为如果第一个请求失败重试，第二个请求可能先成功。

• 需要严格保证分区内顺序的场景（如订单状态变更）：在启用幂等性时，Kafka可以保证顺序；若不启用幂等性，则需将此参数设为1。
• 对顺序不敏感、追求高吞吐的场景：可以保持默认值5或适当调高。

retries与retry.backoff.ms生产者发送失败后的重试机制。retries默认值为Integer.MAX_VALUE，配合delivery.timeout.ms（默认2分钟）一起工作。retry.backoff.ms是重试间隔。对于关键业务，建议保留默认的重试逻辑，但务必设置合理的delivery.timeout.ms，并监听发送回调（Callback）以处理最终失败的消息。

3. 消费者配置：拉取、提交与均衡的艺术

如果说生产者是“推”，那么消费者就是“拉”。消费者的核心挑战在于：如何高效、可靠地从分区拉取数据，并管理消费进度（偏移量），同时还要优雅地应对消费者组内实例的增减（再均衡）。

3.1 心跳、拉取与提交：维持消费生命线的三要素

session.timeout.ms与heartbeat.interval.ms：证明自己还“活着”消费者通过定期向Broker发送心跳来表明自己属于某个消费者组且健康。

• session.timeout.ms： Broker认为消费者失效的超时时间，默认45秒。如果在此时间内未收到消费者的心跳，Broker会将其踢出组，触发再均衡。设置过短容易因GC停顿或网络波动导致误判；设置过长则意味着故障消费者被发现的时间变长，期间其负责的分区无法被消费。
• heartbeat.interval.ms： 消费者发送心跳的频率，默认3秒。通常设置为session.timeout.ms的三分之一左右。例如，session.timeout.ms=30000（30秒），则heartbeat.interval.ms=10000（10秒）。

注意： 新版Kafka将session.timeout.ms的默认值降到了10秒，以加快再均衡速度，但对应用的健康状态要求更高。

max.poll.interval.ms：处理能力的宣言这是最容易引发问题的参数之一，默认5分钟。它定义了消费者调用poll()方法的最大间隔时间。如果两次poll()的间隔超过此值，Broker会认为消费者处理能力不足或已僵死，将其踢出组，触发再均衡。核心逻辑：poll()方法不仅拉取数据，它还负责向Broker发送心跳。如果你在poll()之后的消息处理逻辑非常耗时（比如复杂的计算、同步调用外部API、长时间的数据库事务），就必须调大这个参数，或者将处理逻辑异步化，确保能定期调用poll()。

// 错误示例：处理逻辑耗时过长，可能导致被误踢 while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { // 假设这个process函数执行需要2分钟 timeConsumingProcess(record); // 如果处理一批记录总时间超过 max.poll.interval.ms， 消费者会被踢出 } } // 改进思路：异步处理或调整参数 props.put("max.poll.interval.ms", "300000"); // 调大为5分钟 // 并且确保 process 逻辑不会阻塞 poll 循环太久

fetch.min.bytes与max.poll.records：控制拉取的“胃口”

• fetch.min.bytes： 消费者一次拉取请求期望的最小数据量，默认1字节。Broker会等待有足够的数据后再返回响应。适当调大（如设为1KB或5KB）可以减少网络往返和Broker压力，提升吞吐，但会增加延迟。
• max.poll.records： 一次poll()调用返回的最大记录数，默认500。它控制了单次处理的数据量上限，是防止消费者内存溢出和协调处理耗时与max.poll.interval.ms的关键参数。如果单条消息很大，或者处理逻辑很重，应该调小这个值。

3.2 偏移量提交：可靠性的核心

消费者需要告诉Kafka：“我已经处理到这里了”。这个位置就是偏移量（Offset）。提交方式决定了“至少一次”、“至多一次”还是“恰好一次”的消费语义。

• 自动提交 (enable.auto.commit=true)： 默认方式。消费者后台线程定期（auto.commit.interval.ms，默认5秒）提交已拉取消息的偏移量。风险： 如果在提交后、处理完消息前消费者崩溃，消息会丢失（因为偏移量已向前移动）。如果在处理完消息后、自动提交前崩溃，消息会被重复消费。
• 手动同步提交 (commitSync())： 在处理完一批消息后，手动调用consumer.commitSync()。这保证了处理完才提交，是“至少一次”语义的常用方式。缺点是同步操作会阻塞，影响吞吐。
• 手动异步提交 (commitAsync())： 调用consumer.commitAsync()，不会阻塞。性能更好，但提交失败时不会自动重试，通常需要配合回调函数记录错误或进行重试。
• 更精细的手动提交： 可以提交特定的偏移量，例如在处理每条消息后立即提交其偏移量，但这会严重降低性能。

推荐模式：手动提交 + 同步异步结合

Properties props = new Properties(); props.put("bootstrap.servers", "kafka1:9092,kafka2:9092"); props.put("group.id", "my-consumer-group"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // 关闭自动提交，采用手动控制 props.put("enable.auto.commit", "false"); // 调整拉取和心跳参数 props.put("session.timeout.ms", "30000"); props.put("heartbeat.interval.ms", "10000"); props.put("max.poll.interval.ms", "300000"); props.put("max.poll.records", "100"); // 根据处理能力调整 KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList("my-topic")); try { while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000)); for (ConsumerRecord<String, String> record : records) { // 处理消息 processMessage(record); } // 批量处理完成后，先尝试异步提交（性能好） consumer.commitAsync((offsets, exception) -> { if (exception != null) { log.error("异步提交失败，偏移量: {}", offsets, exception); // 这里可以加入重试逻辑，例如将失败的偏移量存入DB，由后台线程重试 } }); // 为了更强的保证，可以在循环若干次后，或在关闭消费者前，进行一次同步提交兜底 // consumer.commitSync(); } } catch (Exception e) { log.error("消费过程发生异常", e); } finally { try { // 退出前尝试一次同步提交，确保偏移量不丢失 consumer.commitSync(); } finally { consumer.close(); } }

3.3 再均衡监听器：优雅处理分区分配

当消费者组内成员变化（如实例启动、关闭、崩溃）时，会触发再均衡，分区会被重新分配。如果不做处理，可能会导致重复消费或消息丢失。

ConsumerRebalanceListener接口允许你在再均衡发生前后插入钩子逻辑：

• onPartitionsRevoked: 在分区被收回前调用。这是提交偏移量的最后机会，确保当前处理进度被保存。也可以在这里完成一些清理工作。
• onPartitionsAssigned: 在分区被分配后调用。可以在这里初始化状态，或者从自定义存储（如数据库）中读取偏移量，实现更灵活的位移管理。

consumer.subscribe(Arrays.asList("my-topic"), new ConsumerRebalanceListener() { @Override public void onPartitionsRevoked(Collection<TopicPartition> partitions) { // 1. 停止处理这些分区的消息（如果使用多线程） // 2. 提交偏移量！这是最关键的一步。 consumer.commitSync(currentOffsets); // currentOffsets需要自己维护 log.info("分区被收回: {}", partitions); } @Override public void onPartitionsAssigned(Collection<TopicPartition> partitions) { // 1. 可以在这里从外部存储（如DB）读取偏移量，并使用consumer.seek()定位 // 2. 初始化针对这些分区的处理状态 log.info("获得新分区分配: {}", partitions); } });

4. 实战案例：构建一个高可靠订单状态同步服务

让我们将上述所有配置融入一个实战场景：一个电商系统的订单状态变更同步服务。生产者需要将订单状态变更（如“已支付”、“已发货”）可靠地发送到Kafka，消费者需要准实时地消费这些消息，并更新搜索引擎索引和推送用户通知。

4.1 生产者端配置与代码实现

需求分析： 订单状态是核心业务数据，不允许丢失，顺序性很重要（同一个订单的状态变更必须按顺序处理），同时要保证一定的吞吐量以应对大促。

public class OrderStatusProducer { private final KafkaProducer<String, String> producer; private final String topic; public OrderStatusProducer(String bootstrapServers, String topic) { Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // 核心可靠性配置 props.put(ProducerConfig.ACKS_CONFIG, "all"); // 最强可靠性 props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); // 启用幂等，避免重复和保证顺序 // 启用幂等后，max.in.flight.requests.per.connection 会自动设为5或以下，且retries为Integer.MAX_VALUE // 性能与资源调优 props.put(ProducerConfig.LINGER_MS_CONFIG, 5); // 适当聚合，提升吞吐 props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768); // 32KB批次 props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4"); // LZ4压缩，平衡速度与压缩率 props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); // 32MB缓冲区 // 容错配置 props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120000); // 总发送超时2分钟 props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000); // 单次请求超时30秒 this.producer = new KafkaProducer<>(props); this.topic = topic; } public void sendOrderStatus(String orderId, String status) throws ExecutionException, InterruptedException { // 使用订单ID作为Key，确保同一订单的消息进入同一分区，从而保证分区内顺序 ProducerRecord<String, String> record = new ProducerRecord<>(topic, orderId, status); // 使用带回调的send方法，便于监控和异常处理 producer.send(record, (metadata, exception) -> { if (exception != null) { // 发送失败，记录到死信队列或数据库，触发告警，供后续人工或自动补偿 log.error("订单状态发送失败, orderId: {}, status: {}", orderId, status, exception); // 这里可以接入你的监控和补偿系统 Metrics.counter("kafka.producer.failure").increment(); } else { log.debug("订单状态发送成功, topic: {}, partition: {}, offset: {}", metadata.topic(), metadata.partition(), metadata.offset()); } }); // 如果需要更强的同步保证，可以使用 send().get()，但会牺牲性能 } public void close() { producer.close(Duration.ofSeconds(30)); // 优雅关闭，等待未完成请求 } }

4.2 消费者端配置与代码实现

需求分析： 不能丢失消息（至少一次消费），处理逻辑涉及外部调用（更新ES、发推送），可能耗时，需要能优雅应对再均衡。

public class OrderStatusConsumer { private final KafkaConsumer<String, String> consumer; private final String topic; private final ExecutorService processorPool; // 用于异步处理的线程池 private final Map<TopicPartition, OffsetAndMetadata> currentOffsets = new ConcurrentHashMap<>(); public OrderStatusConsumer(String bootstrapServers, String groupId, String topic) { Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); // 关闭自动提交，采用手动提交 props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); // 会话与心跳 props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000); props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 10000); // 拉取控制：考虑到处理逻辑涉及IO，调大poll间隔，调小单次拉取量 props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000); // 5分钟 props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50); // 每次最多拉50条 props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1024); // 等待至少1KB数据 props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500); // 最长等待500ms // 从最早开始消费，仅当没有提交过偏移量时生效 props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); this.consumer = new KafkaConsumer<>(props); this.topic = topic; this.processorPool = Executors.newFixedThreadPool(10); // 根据实际情况调整线程数 } public void consume() { consumer.subscribe(Arrays.asList(topic), new OrderRebalanceListener()); try { while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000)); if (!records.isEmpty()) { // 使用线程池异步处理，避免阻塞poll循环 processRecordsAsync(records); // 注意：异步处理后，偏移量提交需要更精细的控制，见下文 } } } catch (WakeupException e) { // 忽略，用于关闭消费者 } catch (Exception e) { log.error("消费循环发生未知异常", e); } finally { commitOffsetsSync(); // 最终同步提交一次 consumer.close(); processorPool.shutdown(); } } private void processRecordsAsync(ConsumerRecords<String, String> records) { // 将记录按分区分组，方便按分区提交偏移量（更安全） Map<TopicPartition, List<ConsumerRecord<String, String>>> recordsByPartition = records.partitions().stream() .collect(Collectors.toMap(p -> p, records::records)); for (Map.Entry<TopicPartition, List<ConsumerRecord<String, String>>> entry : recordsByPartition.entrySet()) { TopicPartition partition = entry.getKey(); List<ConsumerRecord<String, String>> partitionRecords = entry.getValue(); // 提交给线程池处理 processorPool.submit(() -> { try { for (ConsumerRecord<String, String> record : partitionRecords) { // 业务处理：更新ES、发送通知等 boolean success = processOrderStatus(record.key(), record.value()); if (success) { // 处理成功，记录待提交的偏移量（这里记录最后一条的offset+1） // 更安全的做法是每条处理成功后都记录，这里简化示例 synchronized (currentOffsets) { currentOffsets.put(partition, new OffsetAndMetadata(record.offset() + 1)); } } else { // 处理失败，记录日志，可以进入死信队列或重试队列 log.error("订单状态处理失败， orderId: {}, status: {}", record.key(), record.value()); // 注意：这里没有更新偏移量，下次poll会再次拉取到这条消息（至少一次语义） } } // 这个分区的一批记录处理完后，尝试异步提交这个分区的偏移量 commitOffsetsAsyncForPartition(partition); } catch (Exception e) { log.error("处理分区 {} 的消息时发生异常", partition, e); } }); } } private void commitOffsetsAsyncForPartition(TopicPartition partition) { OffsetAndMetadata offsetMeta; synchronized (currentOffsets) { offsetMeta = currentOffsets.get(partition); } if (offsetMeta != null) { Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(partition, offsetMeta); consumer.commitAsync(offsets, (offsetsMap, exception) -> { if (exception != null) { log.error("异步提交分区 {} 偏移量失败", partition, exception); // 可以存储到可靠存储，后续恢复 } else { log.debug("分区 {} 偏移量提交成功", partition); // 提交成功后，可以从currentOffsets中移除，防止重复提交（可选） } }); } } private void commitOffsetsSync() { try { Map<TopicPartition, OffsetAndMetadata> offsetsToCommit; synchronized (currentOffsets) { offsetsToCommit = new HashMap<>(currentOffsets); } if (!offsetsToCommit.isEmpty()) { consumer.commitSync(offsetsToCommit); log.info("最终同步提交偏移量成功"); } } catch (Exception e) { log.error("最终同步提交偏移量失败", e); } } // 再均衡监听器实现 private class OrderRebalanceListener implements ConsumerRebalanceListener { @Override public void onPartitionsRevoked(Collection<TopicPartition> partitions) { log.info("再均衡触发，分区将被收回: {}", partitions); // 1. 停止对应分区的处理线程（在实际复杂多线程模型中需要） // 2. 立即同步提交所有已处理的偏移量，这是保证至少一次语义的关键 commitOffsetsSync(); } @Override public void onPartitionsAssigned(Collection<TopicPartition> partitions) { log.info("再均衡完成，获得新分区: {}", partitions); // 可以从外部存储（如数据库）读取偏移量，并使用consumer.seek()定位到指定位置 // 例如：for (TopicPartition tp : partitions) { long storedOffset = getOffsetFromDB(tp); consumer.seek(tp, storedOffset); } } } private boolean processOrderStatus(String orderId, String status) { // 模拟耗时业务处理 try { // 1. 更新Elasticsearch中的订单状态 // updateES(orderId, status); // 2. 发送APP推送 // sendPushNotification(orderId, status); Thread.sleep(100); // 模拟处理耗时 log.info("已处理订单状态, orderId: {}, status: {}", orderId, status); return true; } catch (Exception e) { log.error("处理订单状态异常", e); return false; } } }

这个案例展示了如何将可靠性配置（acks=all, 幂等性, 手动提交）、性能调优（批次、压缩）和容错机制（再均衡监听、异步处理、偏移量管理）结合起来，构建一个适用于关键业务场景的Kafka客户端应用。其中，异步处理结合按分区提交偏移量的模式，既保证了消费吞吐，又最大限度地避免了重复消费。