目录

• 一、从基础的客户端说起
  • 消息发送者主流程
  • 消息使用者主流程
• 二、从客户端属性梳理工作方式
  • 使用者分组消费机制
  • 生产者拦截器机制
  • 消息序列化机制
  • 消息分区路由机制
  • 生产者消息缓存机制
  • 发送应答机制



使用者痛点：

大量配置项使得初学者容易混淆，特别是acks/enable.idempotence/max.in.flight.requests.per.connection等组合使用时错误配置会导致吞吐下降或数据丢失。

如何解决：

先把常用模式（比如acks=all + enable.idempotence=true + max.in.flight.requests.per.connection=5)固定为默认值，接下来根据业务场景再细调。怎么说呢，-->
SpringBoot集成Kafka步骤简述：

一、从基础的客户端说起

Kakfa 提供了比较简单的客户端 API。只需引入一个 Maven 依赖即可：

org.apache.kafka
kafka_2.x
x.x.x

1️⃣ 消息发送者主流程

常见痛点：Topic 未提前创建 → send 报错 “Unknown topic”。

# 建议提前创建 Topic：

bin/kafka-topics.sh --bootstrap-server worker1:9092 --create \
--topic myTopic --partitions 3 --replication-factor 1

# 示例代码:

public static void main throws Exception {
Properties props = new Properties;props.put，props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");KafkaProducer producer = new KafkaProducer<>;// 单向发送
ProducerRecord record =
new ProducerRecord<>,"msg"+i);producer.send;// 同步发送
RecordMetadata meta = producer.send.get;System.out.printf("sent to %s partition %d offset %d%n"。meta.topic,meta.partition,meta.offset);// 异步回调示例
producer.send -> {
if {
System.err.println;} else {
System.out.printf("async sent to %s:%d:%d%n"。metaData.topic,metaData.partition,metaData.offset);}
})，}
producer.close;

# 步骤：

1. 设置主要属性——必填，否则连接不上。
2. 建立 ProducerRecord — key 用于分区；value 为业务数据，
3. 选择发送方式：单向 / 同步 / 异步 —— 根据业务需求平衡吞吐与可靠性。不过，

2️⃣ 消息使用者主流程

# 常见痛点：未设置 group.id → Consumer 会自动开启独占模式。无法实现负载均衡，

public static void main {
Properties props = new Properties;props.put，怎么说呢，props.put;不过，props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG。"org.apache.kafka.common.serialization.StringDeserializer");props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");KafkaConsumer consumer = new KafkaConsumer<>;consumer.subscribe);while {
ConsumerRecords records =
consumer.poll);for {
System.out.printf("offset=%d key=%s value=%s%n"。r.offset,r.key,r.value);}
// 手动提交 offset
consumer.commitSync;}
}
consumer.close;


BROKER 自动重试 → 消费不到所有数据？原因是没有手动提交 offset 或 auto.commit 配置不当。
Ack 设置为0 → 消息可能被丢失，但吞吐最高；其实，推荐至少设置为1或all。IDEMPOTENCE + ACKS=all 能保证 Exactly‑once，但需开启事务并适当处理重试异常。二、从客户端属性梳理工作方式

Kafka 的设计要在网络不稳定、服务随时崩溃等极端场景下保证高吞吐。下面逐个拆解关键属性与其业务意义，并结合痛点提示你如何避免常见坑。


使用者分组消费机制 —  

# 问题：同一个 topic 多个 ConsumerGroup 同时订阅，会造成重复消费。为什么，因为每个 Group 独立维护 offset；Broker 在推送时只针对同一 Group 推送一次副本。


Ack 与 Offset 的同步是关键 —— 如果你忘记提交 offset。则 Broker 会认为该条记录未处理完，将
推送给同一 Group 内其他实例，造成重复消费。.
`auto.offset.reset` 的三种策略——`earliest`/`latest`/`none` —— 当 Group 没有已知 offset 时决定读取位置；若不慎选择 `none` 而且不存在 offset，会抛异常导致应用停滞。.Partitioner 分配策略 —  


`default-partitioner`: 基于 key hash 分配 —— 当 key 为 null 时随机分区；这可能造成热点，方法是合理设计 key 或自定义 Partitioner。.
`RoundRobinPartitioner`: 每条记录轮询不同 Partition，不考虑 key；适合无业务关联度但需要均匀负载时使用。.使用者 Partition 分配策略 —  


`RangeAssignor`: 按顺序划分。一旦发生 rebalance 会完全重新划分，可产生大量重新拉取的数据量。适用于单节点小集群，.

'style='margin-left:20px'>
* `` 默认32MB；若生产负荷极高可调大，以免阻塞。* `` 默认16KB；批量越大，每条请求 overhead 越低，但也更易受延迟影响。* `` 默认0，即立即发出 batch；若想进一步压缩网络请求，可设为几毫秒。## ACK 与事务安全性 —  

'style='margin-left:20px'>
* **ACK=0** — 极限吞吐但无安全性，适用于日志采集等可容忍丢失的数据。* **ACK=1** — Leader 写入确认即可，有一定可靠性。* **ACK=all / -** — 所有 ISR 确认后才返回。最安全，但延迟最高，* 开启 **idempotence=true** 必须配合 `acks=all`,`max.in.flight.requests.per.connection<=5`, `retries>=1`;否则会抛异常，怎么说呢，## 消息压缩技术 —  

'style='margin-left:20px'>
* 四种压缩算法可选：gzip、snappy、lz4、zstd。* zstd 压缩比最高但 CPU 开销大；说起来，lz4 性能最快但压缩率略低。* 若服务器 CPU 已经紧张，请不要开启压缩，否则反而降低吞吐。## 事务与幂等性的组合使用 —  

'style='margin-left:20px'>
* **事务 ID 单一实例唯一** —— 多实例共享同 ID 时新实例会被 fence 掉，从而保证一致性。* **跨会话补齐** —— 若实例崩溃后重启相同事务 ID。新实例会自动完成旧事务补齐，使得旧事务不会“悬空”。话说回来，* 使用事务时请确保：
+ `transactional.id.enabled=true`;+ `max.in.flight.requests.per.connection<=5`;
    + 合理设置 `retries`.
三、SpringBoot 集成 Kafka 简易教程 —  
### 第一步先：Maven 引入依赖

org.springframework.kafka
spring-kafka



接下来这方面，application.properties 中统一配置

...

：使用框架注入 KafkaTemplate 与 @KafkaListener
java
@RestController public class DemoController {
@Autowired private KafkaTemplate kafkaTemplate;@GetMapping
public String send{
kafkaTemplate.send;老实说，return \"OK\";}
}
@Component public class DemoListener {
@KafkaListener
public void handleMsg{
System.out.println;}
}


小结与常见问题解答


基本生产者参数

©©©©©©©©©©©©©
请按实际情况替换
# 注意事项
- 对于高可用场景请务必开启 'acks=all'
- 如果需要 Exactly‑once，请同时开启 'enable-idempotence=true'
- 大文件场景可以考虑增加 'batch-size','linger.ms'。'buffer-memory'
\u00B7\u00B7\u00B7\u00B7\u00B7 \u2026 \u2026 \u2026 \u2026


Key
Value
说明
默认值
建议范围
备注
例子

b​ootstrap‑servers​ 
"kafka-broker1,kafka-broker2"
"kafka broker 地址列表"
"kafka-broker-host:kafka-broker-port"
"kafka‑cluster 内部 IP 通信地址"
"- "
"b​ootstrap‑servers – required“


acks​ 
"-“ | '+’ | '+’| '-’ | '-’ )…，…，…，…| ，…，.. ... | ... | ... | ... |... | ...\r



场景
推荐配置
常见错误




高并发写入而且对数据安全要求高
acks=all,enable.idempotence=true,max.in.flight.requests.per.connection≤5,retries≥10
忘记关闭 auto.commit 导致重复投递


大量日志采集且可容忍丢失
acks=0,buffer.memory增大,linger.ms设为数毫秒
对于短期网络波动导致部分日志丢失


多分区多副本读取性能调整
group.instance.id 为固定值。assignor 为 sticky 或 range
rebalance 导致短暂停顿





提示一下： if you still encounter “Unknown topic” errors after creating topics programmatically or via CLI,double-check that bootstrap servers match actual cluster host/port and that your client has network reachability to those brokers.