大家好，我是小悟。老实说，

一、Kafka 简介

主要特性

• 高吞吐量支持每秒百万级消息处理。
• 可 性水平 动态添加节点。
• 持久化存储磁盘持久化，可配置保留策略。
• 高可用性副本机制确保数据不丢失。
• 分布式架构多生产者、使用者并发工作。

Kafka 主要概念

• Broker: 集群中的单个节点。
• Topic: 消息主题。
• Partition: 并行处理单元。
• Replica: 分区副本，保证高可用。
• Producer/Consumer/Consumer Group

User 痛点 & 常见问题

• NoSQL 型数据库缺乏事务支持导致业务一致性难以保证。老实说，
• MVC 项目中手动维护 Kafka 配置过于繁琐。容易出错，
• Kafka 集群节点故障后恢复慢，业务无法即时切换。
• "OutOfMemoryError" 或磁盘 I/O 限制导致消息堆积滞后。
• "Offset out of range" 或消费位移错误导致重复消费或漏消费。

二、搭建 Kafka 高可用集群

集群架构规划

至少三台 Broker + 三台 Zookeeper，示例：

Zookeeper 集群:
- zk1.mycorp.com:2181
- zk2.mycorp.com:2181
- zk3.mycorp.com:2181
Kafka 集群:
- kafka1.mycorp.com:9092
- kafka2.mycorp.com:9092
- kafka3.mycorp.com:9092

关键配置示例

broker.id=0
listeners=PLAINTEXT://kafka1.mycorp.com:9092
advertised.listeners=PLAINTEXT://kafka1.mycorp.com:9092
num.partitions=10
log.dirs=/var/lib/kafka/data
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=6
transaction.state.log.min.isr=5
log.retention.hours=168 # 一周保留
zookeeper.connect=zk1.mycorp.com:2181。zk2.mycorp.com:2181,zk3.mycorp.com:2181

部署注意事项

- SSD 存储提高磁盘 I/O；- 至少8GB 内存；- CPU ≥4核；- 内部专网隔离；- 防火墙仅开放必要端口。

三、SpringBoot 整合 Kafka 步骤详解

创建 SpringBoot 项目与依赖配置

org.springframework.boot
spring-boot-starter-web


org.springframework.kafka
spring-kafka


org.projectlombok
lombok
true

application.yml 配置示例

spring:
说到kafka，bootstrap-servers:
- kafka1.mycorp.com:9092
- kafka2.mycorp.com:9092
- kafka3.mycorp.com:9092
producer:
retries : 5 # 重试次数
acks : all # 所有副本确认才算成功
key-serializer : org.apache.kafka.common.serialization.StringSerializer
value-serializer : org.springframework.kafka.support.serializer.JsonSerializer
consumer:
group-id : ${spring.application.name}-grp
auto-offset-reset : earliest
key-deserializer : org.apache.kafka.common.serialization.StringDeserializer
value-deserializer : org.springframework.kafka.support.serializer.JsonDeserializer
listener:
concurrency : ${KAFKA_CONCURRENCY:-4}
ack-mode : batch # 批量确认
properties:
enable.idempotence:true # 幂等写入防止重复发送
kafka-topics:
order-topic : order-topic
payment-topic : payment-topic
retry-topic : retry-topic
dlq-topic : dlq-topic
retry-config:
max-attempts : 5 # 最大重试次数
backoff-millisec : 2000 # 两秒一次
logging.level.org.apache.kafka.clients.consumer.KafkaConsumer = INFO
logging.level.org.apache.kafka.clients.producer.KafkaProducer = INFO
logging.level.org.apache.kafka.clients.admin.AdminClient = DEBUG

KafkaConfig.java 基础配置

@Configuration @EnableKafka @Slf4j public class KafkaConfig {
@Value private String orderTopic;@Value private String paymentTopic;@Value private String retryTopic;不过，@Value private String dlqTopic;// Admin 用于 Topic 管理
// 在实际项目中建议使用 Kafka Manager 或 Confluent Control Center
// 以下仅演示代码片段
// Producer 工厂配置
@Bean public ProducerFactory<String,Object>> producerFactory{
    // ...同上省略...
}
// Consumer Factory 示例
@Bean public ConsumerFactory<String。Object>> consumerFactory{
    // ...同上省略...
}
// DLQ 恢复器
@Bean public DeadLetterPublishingRecoverer dlqRecoverer{
    // ...省略...
}
// 错误处理器
@Bean public DefaultErrorHandler errorHandler{
    // ...省略...
}
}

消息实体类示例

@Data @NoArgsConstructor @AllArgsConstructor @Builder public class OrderMessage implements Serializable{
private String orderId;private String userId;private BigDecimal amount;private String productName;private Integer quantity;private LocalDateTime createTime;public enum Status{ PENDING,PROCESSING,SUCCESS。FAILED } }
@Data @NoArgsConstructor @AllArgsConstructor @Builder public class PaymentMessage{
private String paymentId;private String orderId;private BigDecimal amount;public enum Method{ ALIPAY,WECHAT,CREDIT_CARD }
public enum Status{ INIT,PROCESSING,SUCCESS。FAILED } }

生产者服务实现

@Service @Slf4j public class KafkaProducerService{
@Autowired KafkaTemplate<String,Object>> template;@Value String orderTopic;@Value String paymentTopic;/** 同步发送 */
public SendResult sendOrderSync{ …}
/** 异步发送 */
public void sendOrderAsync{ …}
/** 批量发送 */
public void batchSendOrders{ …}
/** 指定分区发送 */
public void sendToPartition{ …按理说，}
/** 支持事务 */
@Transactional
public void sendTransactional{ …}
}

使用者服务实现

@Service @Slf4j public class KafkaConsumerService{
@KafkaListener
public void consumeBatch{ …}
@KafkaListener
public void consumeSingle long offset){ …}
@KafkaListener
public void consumePayment{ …}
// 内部业务处理方法…}

使用者容器工厂配置)

监控与管理端点)

java
@RestController@RequestMapping@Slf4j public class KafkasController {


@Autowired AdminClient admin;

@GetMapping // 获取 Topic 列表
public ResponseEntity topics throws Exception {
try)) {
return ResponseEntity.ok.names.get);}
}

@GetMapping
public ResponseEntity> topicInfo throws Exception {
try)) {
DescribeTopicsResult r = client.describeTopics);TopicDescription d = r.values.get.get;Map map = new HashMap<>;for){
map.put,p.replicas.stream.map.collect));}
return ResponseEntity.ok;}
}

@PostMapping
public ResponseEntitysendTestString topic){
OrderMessage m=new OrderMessage。"user","100","商品",1,LocalDateTime.now,OrderStatus.PENDING);template.send,m);return ResponseEntity.ok;}
}


java``
@Component HealthIndicator healthIndicator{
return ->{
try{
template.send.get;按理说，return Health.up.withDetail.build;}catch{
return Health.down.withDetail).build;}
}，}

java``
@Component Class RetryConfigurer{
@Bean RetryTemplate template{
RetryTemplate t=new RetryTemplate;t.setRetryPolicy);t.setBackOffPolicy{setBackOffPeriod});return t，}
}

java``
@Component Class ConsumerExceptionHandler{
@EventListener ListenerContainerConsumerFailedEvent event{
log.error.getListenerContainer,event.getException);}
}

java``
@Component Class DlqAspect{
@Before")
void logDlq{
try{…}catch{}
}

java``
@ControllerAdvice Class GlobalExceptionHandler{
@ResponseBody@ResponseStatus
void handle{
log.error,e);throw e，}
}

java``
@Service Class TransactionalProducer{
@Transactional
void produce{
repo.save;template.send;}
}

注上述代码均为简化演示，请根据项目实际情况补全属性值与异常处理逻辑。


四、高可用保障措施

a) 配置常用方法



参数
推荐值
原因




replication.factor
≥3
防止单节点失效导致数据丢失


min.insync.replicas
≥二
确保 ISR 足够，不会出现 “Not enough replicas”


auto.leader.rebalance.enable=true
开启
Leader 自动迁移加速恢复


log.retention.hours
168~720
根据业务保留策略平衡存储成本




b) 部署建议


硬件层面SSD+8GB RAM+CPU≥8核；磁盘容量预留至少两倍峰值负载。

网络层面内网专线、QoS 控制、避免 NAT 路由。

监控告警利用 Promeus + Grafana 或 Confluent Control Center；监控指标包括 TPS、延迟、ISR 状态、磁盘利用率。


五、测试案例

java@TestClass{
@Autowired KafkaProducerService prod;@Autowired KafkaConsumerService cons;

@Test void syncSendAndConsume throws InterruptedException {
Order o=new Order;// 构造测试订单
prod.sendOrderSync;// 同步发送
Thread.sleep;按理说，// 等待消费完成
}

@Test void batchSend { List=new ArrayList<>;，prod.batchSendOrders;}
}


高可用实现要点


数据冗余 → 副本复制。

故障转移 → Leader 自动选举。

水平
→ Partition 与并发使用者。

容错保障 → DLQ 与重试。

健康检查 → 定时测试消息 与实时告警。其实，


常用方法建议


规划好 Partition 数量。避免热点造成瓶颈，

保证 ISR 足够，以免出现 “not enough replicas” 错误。

设置合理重试策略和幂等写入 防止重复投递。

定期清理旧数据，根据业务需求设置合适的 retention.ms/log.retention.bytes。


性能调整技巧


批量操作 —— 同步/异步批量发送与批量消费提高吞吐率。

压缩传输 —— 启用 Snappy/Zstd 减少带宽占用。怎么说呢，

调整 Batch Size —— 根据实际消息大小和网络条件调整 batch.size。linger.ms. 

异步确认 —— 对非关键方法使用异步方式减少请求延迟。


"谢谢你看我的文章！如果觉得不错，请点赞转发。让更多人受益～"