企业官网建设流程全解析

Kafka是高吞吐分布式消息系统，广泛用于日志采集、流处理、解耦。本文讲解Kafka核心概念、Spring Boot集成、生产实践。

一、核心概念

• Broker：Kafka节点

• Topic：消息主题

• Partition：分区，并行单位

• Replica：副本

• Producer/Consumer

• Consumer Group

• Offset

二、Spring Boot集成

xml

复制

下载

运行

<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>

yaml

复制

下载

spring: kafka: bootstrap-servers: localhost:9092 producer: key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.springframework.kafka.support.serializer.JsonSerializer acks: all retries: 3 consumer: group-id: my-group auto-offset-reset: earliest enable-auto-commit: false key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer properties: spring.json.trusted.packages: "*"

三、生产者

java

复制

下载

@Service @RequiredArgsConstructor public class OrderProducer { private final KafkaTemplate<String, Order> kafkaTemplate; public void send(Order order) { kafkaTemplate.send("order-topic", order.getId().toString(), order) .whenComplete((result, ex) -> { if (ex != null) { log.error("发送失败", ex); } else { log.info("发送成功: {}", result.getRecordMetadata().offset()); } }); } }

四、消费者

java

复制

下载

@Component @Slf4j public class OrderConsumer { @KafkaListener(topics = "order-topic", groupId = "order-group", concurrency = "3") public void consume(ConsumerRecord<String, Order> record, Acknowledgment ack) { try { log.info("收到订单: {}", record.value()); processOrder(record.value()); ack.acknowledge(); } catch (Exception e) { log.error("处理失败", e); // 不ack，触发重试 } } }

手动提交offset：

yaml

复制

下载

spring: kafka: consumer: enable-auto-commit: false listener: ack-mode: manual_immediate

五、分区与顺序

• 相同key到相同分区

• 单分区内有序

• 需要全局有序：单分区（牺牲并行）

java

复制

下载

kafkaTemplate.send("topic", order.getUserId().toString(), order);

六、消息可靠性

6.1 生产者不丢

• acks=all

• retries>0

• enable.idempotence=true

6.2 Broker不丢

• replication.factor >= 3

• min.insync.replicas >= 2

• unclean.leader.election.enable=false

6.3 消费者不丢

• 处理成功后手动提交

• 避免自动提交

七、重复消费与幂等

Kafka保证at-least-once，需业务幂等：

java

复制

下载

public void consume(Order order) { if (redisTemplate.opsForValue() .setIfAbsent("consumed:" + order.getId(), "1", 24, TimeUnit.HOURS)) { processOrder(order); } }

八、死信队列

java

复制

下载

@Bean public DefaultErrorHandler errorHandler(KafkaTemplate<String, Object> template) { DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template); return new DefaultErrorHandler(recoverer, new FixedBackOff(1000, 3)); }

九、性能优化

1. 批量发送：batch.size、linger.ms

2. 压缩：compression.type=lz4

3. 增加分区数

4. 消费者并发：concurrency

5. 手动提交，减少提交频率

十、总结

Kafka是高性能消息队列的首选。掌握分区、副本、offset、幂等，能构建可靠的消息系统。