Kafka KRaft 实践 Spring Boot 微服务

Spring Boot 微服务中的 Kafka 实际实现

中高级开发者可以使用 KRaft 快速部署 Kafka 集群，并将其与在线商店的四个 Spring Boot 服务集成。order-service 通过 REST API 接受订单并发布订单已下事件。inventory-service 预留库存并生成库存已预留或库存拒绝事件。analytics-service 和 notification-service 异步处理这些事件。

要求：Java、Spring Boot、Docker、Docker Compose、Postman。依赖：Spring for Apache Kafka、Spring Web。

Kafka 集群的 Docker Compose

部署三个 Kafka 8.1.0 代理，使用 KRaft 而无需 Zookeeper。每个代理同时扮演代理和控制器角色。在端口 8086 上的 Kafka-UI 用于监控主题、分区和消费者组。

创建 docker-compose.yaml：

services:
  kafka-0:
    image: confluentinc/cp-kafka:8.1.0
    container_name: kafka-0
    hostname: kafka-0
    ports:
      - "29092:29092"
    environment:
      CLUSTER_ID: ZETVJENWRjaECDUEExP_eg
      KAFKA_NODE_ID: 0
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: INTERNAL://:9091,EXTERNAL://:29092,CONTROLLER://:9093
      KAFKA_CONTROLLER_QUORUM_VOTERS: 0@kafka-0:9093,1@kafka-1:9093,2@kafka-2:9093
      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka-0:9091,EXTERNAL://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
  # Similarly for kafka-1 (port 29093), kafka-2 (port 29094)
  kafka-ui:
    image: provectuslabs/kafka-ui:v0.7.2
    container_name: kafka-ui
    ports:
      - "8086:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka-0:9091,kafka-1:9091,kafka-2:9091
      DYNAMIC_CONFIG_ENABLED: true

启动命令：docker compose up -d。集群已准备好加载，具有 3 个分区和 3 倍复制。

Spring Boot 服务配置

通用生产者设置

所有服务的 application.properties：

spring.kafka.bootstrap-servers=localhost:29092,localhost:29093,localhost:29094
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

程序化主题创建

config 包中的 KafkaConfig 类：

@Configuration
public class KafkaConfig {
    @Bean
    public NewTopic orderPlacedTopic() {
        return TopicBuilder.name("order-placed")
            .partitions(3)
            .replicas(3)
            .config(TopicConfig.RETENTION_MS_CONFIG, "86400000")
            .config(TopicConfig.RETENTION_BYTES_CONFIG, "524288000")
            .build();
    }
}

保留期：每个主题 24 小时，500MB。在 inventory-service 中类似处理 inventory-reserved、inventory-rejected。

订单服务实现（生产者）

DTO 和事件

public record OrderRequest(String email, String productName, Integer quantity) {}

public record OrderPlacedEvent(String orderId, String email, String productName, Integer quantity) {}

业务逻辑

@Service
public class OrderService {
    private final KafkaTemplate<String, OrderPlacedEvent> kafkaTemplate;

    public OrderService(KafkaTemplate<String, OrderPlacedEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public String placeOrder(OrderRequest orderRequest) {
        String orderId = UUID.randomUUID().toString();
        OrderPlacedEvent event = new OrderPlacedEvent(orderId, orderRequest.email(), 
            orderRequest.productName(), orderRequest.quantity());
        kafkaTemplate.send("order-placed", event);
        return orderId;
    }
}

REST 控制器立即返回 201 Created 以及 orderId，处理异步进行。

Inventory-Service：生产者 + 消费者

监听 order-placed，检查库存水平，并发布 inventory-reserved 或 inventory-rejected。需要 @KafkaListener 和第二个 KafkaTemplate。

服务角色：

• 纯生产者：order-service
• 纯消费者：analytics-service、notification-service
• 生产者+消费者：inventory-service

消费者实现（analytics、notification）

@KafkaListener(topics = "inventory-reserved")
public void handleInventoryReserved(InventoryReservedEvent event) {
    // Analytics logic
}

Spring Kafka 自动管理偏移量，消费者组由服务名称形成。

测试流程

• POST /orders，JSON { "email": "[email protected]", "productName": "laptop", "quantity": 1 }
• 在 Kafka-UI 中监控：order-placed → inventory-reserved
• analytics 和 notification 并行接收事件

关键点：

• KRaft 无需 Zookeeper 简化部署
• 3 个分区 + 3 倍复制 = 容错性
• 值使用 JsonSerializer，键使用 StringSerializer
• 24 小时/500MB 保留期适合开发
• KafkaTemplate 抽象底层 API

— Editorial Team