kafka

08. Java Spring Kafka 实战

Spring Kafka 配置、@KafkaListener、并发消费者、事务与错误处理

Leeting Yan 2026-08-14 13 分钟阅读 6281 字

Spring Kafka 是 Spring 生态中连接 Kafka 的核心组件，它以声明式的方式将消息生产者和消费者融入 Spring Boot 应用。从简单的消息收发，到事务性消息处理、消费重试、死信队列，Spring Kafka 提供了全方位的企业级能力。本文覆盖 Spring Kafka 的核心配置、生产消费模型、错误处理、并发消费和事务支持，同时深入探讨生产者进阶用法、消费者组再平衡机制、Exactly-Once 语义实现、Kafka Streams 集成、监控运维以及 Schema Registry 等高级主题。

1. 基础配置

Spring Boot 通过 spring.kafka.* 属性实现声明式配置，开发者无需手动编写繁琐的 Properties 对象，即可快速搭建生产级 Kafka 客户端。完整的配置可以覆盖生产者、消费者、监听器以及底层网络参数，满足绝大部分生产场景的需求。

spring:
  kafka:
    bootstrap-servers: kafka1:9092,kafka2:9092,kafka3:9092
    producer:
      acks: all
      retries: 3
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      properties:
        enable.idempotence: true
        max.in.flight.requests.per.connection: 5
        compression.type: zstd
    consumer:
      group-id: order-service
      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:
        isolation.level: read_committed
        spring.json.trusted.packages: "com.example.model"
    listener:
      ack-mode: manual_immediate  # 手动提交模式
      concurrency: 3              # 并发消费者数
      poll-timeout: 3000
      missing-topics-fatal: false

在上述配置中，acks: all 确保消息在所有副本确认后才返回成功，保证最高级别持久性。enable.idempotence: true 开启幂等生产者，自动去重以避免因网络重试导致的消息重复。compression.type: zstd 选用 Zstandard 压缩算法，在吞吐和压缩比之间取得良好平衡。消费者端 enable-auto-commit: false 关闭自动提交，交由业务层在消息处理成功后手动确认，确保消息不会丢失且仅处理一次。isolation.level: read_committed 使消费者仅读取已提交事务范围内的消息，避免读到未提交的事务数据。

2. 生产消息

基础发送

KafkaTemplate 是 Spring Kafka 提供的核心模板类，封装了底层 KafkaProducer 的复杂度，支持同步发送、异步回调和事务发送三种模式。基础用法如下所示：

@Service
public class OrderEventPublisher {
    private final KafkaTemplate<String, Order> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, Order> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void publishOrderCreated(Order order) {
        kafkaTemplate.send("orders.created", order.getUserId(), order)
            .whenComplete((metadata, exception) -> {
                if (exception != null) {
                    log.error("Failed to publish order {}", order.getId(), exception);
                } else {
                    log.info("Published to partition {} offset {}",
                        metadata.partition(), metadata.offset());
                }
            });
    }
}

send() 方法返回 ListenableFuture 对象，通过 whenComplete 注册回调即可在发送完成后获得分区号、偏移量等元数据。异步回调不会阻塞业务线程，非常适合高吞吐场景。如果需要同步等待，可以调用 future.get() 方法，但需注意超时设置以避免线程长时间阻塞。

Kafka 消息头（Headers）机制允许在消息体之外附加键值对元数据，常用于传递跟踪 ID、事件类型、业务标签等上下文信息，便于链路追踪和多租户路由。

public void publishWithHeaders(Order order) {
    ProducerRecord<String, Order> record = new ProducerRecord<>(
        "orders.created",
        order.getUserId(),
        order
    );
    record.headers().add("event-type", "ORDER_CREATED".getBytes());
    record.headers().add("trace-id", MDC.get("traceId").getBytes());

    kafkaTemplate.send(record);
}

通过在 ProducerRecord 中设置 userId 作为消息的 key，可以确保同一用户的所有订单消息落入同一个分区，从而保证该用户消息的消费顺序性。Header 中的 trace-id 可与分布式追踪系统（如 Zipkin、Jaeger）对接，实现端到端的链路追踪。

事务性发送

Kafka 事务支持在同一事务边界内发送多条消息，保证它们要么全部成功提交，要么全部回滚，不会出现部分消息到达而部分丢失的情况。结合 @Transactional 注解，开发者可以用熟悉的事务编程模型编写 Kafka 业务代码。

@Service
public class OrderSagaPublisher {
    private final KafkaTemplate<String, Object> kafkaTemplate;

    @Transactional("kafkaTransactionManager")
    public void publishOrderSaga(Order order) {
        kafkaTemplate.send("orders.created", order.getUserId(), order);
        kafkaTemplate.send("inventory.reserved", order.getId(), new InventoryEvent(order));
        kafkaTemplate.send("payments.charged", order.getId(), new PaymentEvent(order));
        // 三笔消息在一个事务中原子提交
    }
}

事务型生产者需要在配置中指定 transactional-id，且事务 ID 在集群中应全局唯一。Spring Kafka 会自动管理事务的开启、提交和回滚，当事务中任何一步抛出异常时，所有已发送的消息都会被撤销。

3. 消费消息

@KafkaListener 基础用法

@KafkaListener 是 Spring Kafka 提供的核心注解，通过简单的声明即可将普通 Java 方法转化为 Kafka 消费者。它支持灵活的消息参数注入，包括消息体、消息头、分区信息以及手动确认对象。

@Component
public class OrderEventConsumer {

    @KafkaListener(
        topics = "orders.created",
        groupId = "order-service",
        concurrency = "3"
    )
    public void onOrderCreated(
            @Payload Order order,
            @Header("event-type") String eventType,
            Acknowledgment ack) {

        log.info("Received order: {}, event-type: {}", order.getId(), eventType);
        processOrder(order);
        ack.acknowledge();  // 手动确认
    }
}

@Payload 用于注入反序列化后的消息体，@Header 用于提取特定的消息头。Acknowledgment 对象在 ack-mode 为手动模式时自动注入，调用 acknowledge() 即向 Kafka 提交当前消息的偏移量。

批量消费

当单条消息处理开销较低时，批量消费可以显著提升吞吐量。Spring Kafka 通过设置 batch = "true" 启用批量模式，消费者方法参数自动接收一个消息列表。

@KafkaListener(
    topics = "orders.created",
    groupId = "batch-order-service",
    batch = "true",
    concurrency = "2"
)
public void onBatchOrders(List<Order> orders) {
    log.info("Processing batch of {} orders", orders.size());
    orderRepository.saveAll(orders);
}

批量消费时，建议配合 max.poll.records 参数控制每次拉取的消息数量，避免一次拉取过多导致内存溢出或处理超时。同时，批量写入数据库可以大幅减少网络往返和事务开销，是提升消费端性能的重要手段。

消费过滤

Spring Kafka 支持通过 topicPartitions 精确控制监听的分区和偏移量，消费者只处理特定分区的消息，灵活性极高。这在数据迁移、分区级灰度测试等场景中非常有用。

@KafkaListener(
    topics = "orders.*",
    groupId = "filtered-service",
    topicPartitions = {
        @TopicPartition(
            topic = "orders.created",
            partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "0")
        )
    }
)
public void onSpecificPartition(ConsumerRecord<String, Order> record) {
    // 只消费特定分区的消息
}

4. 错误处理与重试

全局错误处理器

在分布式消息系统中，消费过程难免遇到网络超时、数据库死锁等暂时性异常。Spring Kafka 通过 DefaultErrorHandler 提供统一的错误处理机制，支持指数退避重试、异常分类处理和死信队列投递。

@Configuration
public class KafkaErrorConfig {

    @Bean
    public DefaultErrorHandler errorHandler() {
        // 指数退避重试策略
        BackOff fixedBackOff = new ExponentialBackOff(1000, 2);
        ((ExponentialBackOff) fixedBackOff).setMaxInterval(30000);

        DefaultErrorHandler handler = new DefaultErrorHandler(fixedBackOff);

        // 不重试的异常（直接入死信队列）
        handler.addNotRetryableExceptions(
            IllegalArgumentException.class,
            ValidationException.class
        );

        // 自定义死信队列
        DeadLetterPublishingRecoverer recoverer =
            new DeadLetterPublishingRecoverer(kafkaTemplate);
        handler.setBackOffFunction((rec, ex) -> {
            if (ex instanceof DatabaseException) {
                return new FixedBackOff(5000, 3); // DB 异常：重试 3 次，间隔 5s
            }
            return fixedBackOff;
        });

        return handler;
    }
}

通过 addNotRetryableExceptions，可以将格式错误、参数校验失败等不可逆异常直接投递到死信队列，避免无效重试占用系统资源。setBackOffFunction 允许针对不同异常类型定制重试策略，例如数据库异常采用固定间隔重试，而服务端异常采用指数退避。

死信队列（DLQ）

死信队列用于存放超过最大重试次数仍无法处理的消息，是保障系统可用性的关键机制。通过 DeadLetterPublishingRecoverer，Spring Kafka 可以在消息进入死信队列时保留原始分区和偏移量，便于后续人工排查或自动补偿。

@Bean
public DefaultErrorHandler deadLetterErrorHandler(
        KafkaOperations<String, Object> template) {

    DeadLetterPublishingRecoverer recoverer =
        new DeadLetterPublishingRecoverer(template,
            (record, exception) -> {
                // 自定义 DLQ 路由：原始 topic + .dlq 后缀
                return new TopicPartition(record.topic() + ".dlq", record.partition());
            });

    return new DefaultErrorHandler(recoverer, new FixedBackOff(2000, 3));
}

建议为死信队列独立配置消费者，对其进行告警通知、日志记录和自动重处理，形成完整的错误闭环。

5. 并发消费模型

Spring Kafka 的 concurrency 参数控制消费者数量，但需注意它与 Kafka Partition 的关系。Kafka 的设计决定了同一消费组中，一个分区只能被一个消费者实例消费，因此并发消费者数量受限于分区数量。

@KafkaListener(
    topics = "orders.created",
    groupId = "order-service",
    concurrency = "6"  // 最多 6 个并发消费者
)
public void onOrderCreated(Order order) {}

Partition 数	concurrency	实际并发数	说明
3	3	3	完全对齐
3	6	3	3 个消费者空闲
6	3	3	每个消费者处理 2 个分区
6	6	6	完全对齐

最佳实践：concurrency == partition 数量 或 concurrency = partition / N（按需扩展）。

在设计高并发系统时，应根据业务峰值吞吐量预先评估并创建足够的分区数量。分区过少会导致消费端无法水平扩展，分区过多则会增加 Broker 的资源消耗和协调开销。通常建议在初期按数据量创建适量分区，后续通过 Kafka Admin API 动态扩容。

6. 事务集成：Kafka + JDBC

Spring Kafka 支持 Kafka 事务与 JDBC 事务的桥接（Chained Kafka Transaction Manager），可以将数据库操作和消息发送纳入同一个分布式事务。当事务提交时，数据库变更和 Kafka 消息同时对外可见；当事务回滚时，两者一同撤销，彻底避免数据不一致问题。

@Configuration
public class KafkaTxConfig {

    @Bean
    public ChainedKafkaTransactionManager<Object, Object> chainedTxManager(
            KafkaTransactionManager<Object, Object> kafkaTxManager,
            PlatformTransactionManager dataSourceTxManager) {
        return new ChainedKafkaTransactionManager<>(kafkaTxManager, dataSourceTxManager);
    }
}

@Service
public class OrderService {
    @Transactional("chainedTxManager")
    public void createOrder(OrderDTO dto) {
        // 1. 写入数据库
        Order order = orderRepository.save(new Order(dto));

        // 2. 发送 Kafka 消息
        kafkaTemplate.send("orders.created", order.getUserId(), order);

        // 两个操作在同一个分布式事务中
        // 任一失败都会回滚
    }
}

链式事务管理器的执行顺序与传入参数一致。如果数据库事务失败，Kafka 事务不会提交；反之亦然。但需要注意，Kafka 事务基于两阶段提交协议，性能开销高于普通发送，应仅在强一致性场景中使用。

7. Spring Kafka 生产者进阶

KafkaTemplate 异步发送与回调

除了基础的 whenComplete 回调，KafkaTemplate 还支持发送监听器（ProducerListener），实现对发送成功和失败的统一拦截与处理，适用于全局日志记录、指标统计等横切关注点。

@Component
public class LoggingProducerListener implements ProducerListener<String, Object> {

    @Override
    public void onSuccess(ProducerRecord<String, Object> record, RecordMetadata metadata) {
        log.info("Message sent to {}-{} offset {}",
            metadata.topic(), metadata.partition(), metadata.offset());
    }

    @Override
    public void onError(ProducerRecord<String, Object> record, RecordMetadata metadata, Exception exception) {
        log.error("Failed to send message to {}: {}", record.topic(), exception.getMessage());
    }
}

@Configuration
public class KafkaProducerConfig {

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate(ProducerFactory<String, Object> factory) {
        KafkaTemplate<String, Object> template = new KafkaTemplate<>(factory);
        template.setProducerListener(new LoggingProducerListener());
        return template;
    }
}

生产者拦截器

Kafka 客户端支持通过 ProducerInterceptor 在消息发送前和确认后插入自定义逻辑，例如审计日志、消息加密或动态路由。

public class AuditProducerInterceptor implements ProducerInterceptor<String, Order> {

    @Override
    public ProducerRecord<String, Order> onSend(ProducerRecord<String, Order> record) {
        record.headers().add("sent-at",
            Instant.now().toString().getBytes(StandardCharsets.UTF_8));
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        // ACK 回调
    }

    @Override
    public void close() {}

    @Override
    public void configure(Map<String, ?> configs) {}
}

消息 Key 分区策略自定义

默认情况下，Kafka 根据消息 key 的哈希值选择分区。但在某些业务场景下，可能需要自定义分区策略，例如按地理区域路由或实现负载均衡。

public class RegionPartitioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();
        String region = key.toString().split(":")[0]; // key 格式: region:userId
        return Math.abs(region.hashCode()) % numPartitions;
    }

    @Override
    public void close() {}

    @Override
    public void configure(Map<String, ?> configs) {}
}

在 Spring Boot 配置中指定自定义分区器：

spring:
  kafka:
    producer:
      properties:
        partitioner.class: com.example.kafka.RegionPartitioner

8. 消费者组与分区分配

Consumer Group Rebalance 机制

当消费者组中的成员发生变化时，例如新消费者加入、现有消费者退出或主题分区数量变更，会触发再平衡（Rebalance）过程。再平衡期间，整个消费组会暂停消费，直到分区重新分配完成。频繁的再平衡会严重影响消费延迟，因此理解并优化再平衡行为至关重要。

消费者组通过消费协调器（Group Coordinator）与 Broker 通信，Group Coordinator 负责维护组成员关系和分区分配方案。消费者在启动时向 Coordinator 注册，session.timeout.ms 决定消费者被认为失效的时间阈值，heartbeat.interval.ms 控制心跳发送频率。

分区分配策略

Kafka 提供了多种分区分配策略，默认使用 Range，但可以根据业务特点选择更合适的策略。

@Configuration
public class ConsumerAssignmentConfig {

    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-service");
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
            CooperativeStickyAssignor.class.getName());
        // 可选: RangeAssignor, RoundRobinAssignor, StickyAssignor
        return new DefaultKafkaConsumerFactory<>(props);
    }
}

Range：将分区按范围分配给各消费者，分配结果可能不均匀，但内存局部性较好。
RoundRobin：按轮询方式循环分配分区，分配最均匀，但再平衡时可能导致大量分区迁移。
Sticky：优先保持现有分配关系，仅在需要时才迁移分区，兼顾均匀性和迁移最小化。
CooperativeSticky：Kafka 2.4 引入的改进策略，采用增量再平衡，减少再平衡期间的消费暂停时间。

静态成员避免再平衡

Kafka 2.3 引入的静态成员（Static Membership）机制允许消费者重启后保留原有的分区分配，从而避免因短暂重启导致的再平衡。实现静态成员需要设置消费者实例 ID，并且消费者每次重启时使用相同的 ID。

@Bean
public ConsumerFactory<String, Object> staticMemberFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-service");
    props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, "order-service-instance-01");
    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 45000);
    return new DefaultKafkaConsumerFactory<>(props);
}

静态成员适合部署在容器平台（如 Kubernetes）上的场景，当 Pod 因滚动升级或节点迁移而重启时，只要实例 ID 不变，就不会触发不必要的再平衡。

9. Exactly-Once 语义实现

Kafka 的 Exactly-Once 语义（EOS）是流处理系统中最严格的消息保证，要求消息既不会丢失也不会重复。实现 EOS 需要同时满足三个条件：幂等生产者、事务支持以及消费者隔离级别配置。

幂等 Producer + 事务配置

spring:
  kafka:
    producer:
      transactional-id: order-service-producer
      properties:
        enable.idempotence: true
        acks: all
        retries: Integer.MAX_VALUE
        max.in.flight.requests.per.connection: 5

幂等生产者通过为每个生产的消息分配序列号，自动识别并去重网络重试导致的重复消息。开启幂等性时，retries 可安全设置为无限重试，max.in.flight.requests.per.connection 在 Kafka 2.5 之后可设为最多 5 而不影响顺序保证。

Spring Kafka @KafkaListener 事务

@Component
public class ExactlyOnceConsumer {

    @Autowired private KafkaTemplate<String, Object> kafkaTemplate;

    @KafkaListener(topics = "orders.incoming", groupId = "order-processor")
    @Transactional("kafkaTransactionManager")
    public void processOrder(@Payload Order order, Acknowledgment ack) {
        // 处理订单
        orderService.process(order);

        // 转发到下游主题
        kafkaTemplate.send("orders.processed", order.getId(), order);

        // 消费者偏移量与下游发送在同一个事务中
        ack.acknowledge();
    }
}

在此模式下，消费者偏移量的提交与消息处理结果被绑定到同一个事务中。如果消息处理成功但发送下游失败，偏移量不会提交，消息会被重新消费；如果发送成功但处理失败，消息也不会被确认，杜绝了数据丢失或重复的可能性。

Outbox 模式实现

当业务系统同时需要更新数据库和发送 Kafka 消息时，传统事务方案依赖于链式事务管理器，但性能和可用性受限。Outbox 模式通过将消息先写入数据库的本地 Outbox 表，再由单独的转发进程投递到 Kafka，实现了数据库事务与消息发送的解耦。

@Entity
@Table(name = "outbox_events")
public class OutboxEvent {
    @Id @GeneratedValue
    private Long id;
    private String aggregateType;
    private String aggregateId;
    private String eventType;
    private String payload;
    private Instant createdAt;
    private boolean published;
}

@Service
public class OrderService {
    @Transactional
    public void createOrder(OrderDTO dto) {
        Order order = orderRepository.save(new Order(dto));

        OutboxEvent event = new OutboxEvent();
        event.setAggregateType("Order");
        event.setAggregateId(order.getId());
        event.setEventType("ORDER_CREATED");
        event.setPayload(objectMapper.writeValueAsString(order));
        event.setCreatedAt(Instant.now());
        event.setPublished(false);
        outboxRepository.save(event);
    }
}

@Component
public class OutboxRelay {
    @Scheduled(fixedRate = 5000)
    public void relayPendingEvents() {
        List<OutboxEvent> pending = outboxRepository.findByPublishedFalseOrderByCreatedAtAsc();
        for (OutboxEvent event : pending) {
            kafkaTemplate.send(
                "orders.created",
                event.getAggregateId(),
                event.getPayload());
            event.setPublished(true);
            outboxRepository.save(event);
        }
    }
}

Outbox 模式的优势在于不依赖分布式事务，仅靠本地数据库事务即可保证数据一致性；缺点是引入了一个额外的轮询进程，需要额外处理并发投递和幂等性。配合唯一业务键和数据库乐观锁，可以有效防止重复投递。

10. Kafka Streams 集成

Kafka Streams 是 Kafka 原生的流处理库，Spring Cloud Stream 通过 Kafka Streams Binder 将其与 Spring Boot 深度整合，开发者可以使用声明式注解构建实时流处理拓扑。

Spring Cloud Stream Kafka Streams 绑定器

spring:
  cloud:
    stream:
      bindings:
        input-in-0:
          destination: orders.created
        output-out-0:
          destination: orders.processed
      kafka:
        streams:
          binder:
            brokers: kafka:9092
            configuration:
              default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
              default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
              commit.interval.ms: 1000
          bindings:
            input-in-0:
              consumer:
                application-id: order-stream-processor

@Component
public class OrderStreamProcessor {

    @Bean
    public Function<KStream<String, Order>, KStream<String, ProcessedOrder>> processOrders() {
        return input -> input
            .filter((key, order) -> order.getAmount().compareTo(BigDecimal.ZERO) > 0)
            .mapValues(order -> new ProcessedOrder(order))
            .peek((key, processed) -> log.info("Processed order: {}", key));
    }
}

KTable vs KStream

KStream 表示一个无界的事件流，每条数据都是一个独立的事件，消费的是完整的追加日志。
KTable 表示一个可更新的状态表，消费的是 changelog 流，相同 key 的后续消息会覆盖前值。

KTable 适合表示实体状态，KStream 适合表示事件序列，两者可以通过 leftJoin、innerJoin 进行组合运算。

Windowed Aggregation 与 State Store

@Bean
public Function<KStream<String, Order>, KStream<String, BigDecimal>> orderAggregation() {
    return input -> input
        .groupByKey()
        .windowedBy(TimeWindows.of(Duration.ofMinutes(5)).grace(Duration.ofMinutes(1)))
        .aggregate(
            BigDecimal::zero,
            (key, order, total) -> total.add(order.getAmount()),
            Materialized.<String, BigDecimal, WindowStore<Bytes, byte[]>>as("order-totals")
                .withKeySerde(Serdes.String())
                .withValueSerde(new JsonSerde<>(BigDecimal.class))
        )
        .toStream()
        .map((windowedKey, total) ->
            new KeyValue<>(windowedKey.key(), total));
}

窗口聚合允许在滚动时间窗口内对数据进行分组统计，State Store 用于持久化中间状态，支持故障恢复时从最近检查点重建。通过 Interactive Query，还可以暴露 REST 接口查询当前窗口内的聚合值，实现实时仪表盘功能。

11. 监控与运维

Micrometer + Prometheus 暴露 Kafka Metrics

Spring Boot Actuator 配合 Micrometer 可以将 Kafka 客户端的内部指标自动导出为 Prometheus 格式，包括消费者延迟（Consumer Lag）、生产者吞吐量、重平衡频率等关键指标。

@Configuration
public class KafkaMetricsConfig {

    @Bean
    public MeterRegistryCustomizer<MeterRegistry> kafkaMetricsCustomizer(
            KafkaListenerEndpointRegistry registry) {
        return r -> {
            for (MessageListenerContainer container : registry.getListenerContainers()) {
                container.metrics().forEach((clientId, metrics) ->
                    metrics.forEach((name, metric) ->
                        Gauge.builder("kafka." + name.name(), metric, m -> m.metricValue().doubleValue())
                            .tag("client.id", clientId)
                            .tag("group", container.getGroupId())
                            .register(r)
                    )
                );
            }
        };
    }
}

消费者延迟是运维监控中最重要的指标之一。当消费速度低于生产速度时，延迟会持续增加，因此需要配置告警规则，当 lag 超过阈值时触发扩容或通知。

Kafka AdminClient 动态管理 Topic

Spring Kafka 提供了 KafkaAdmin 封装，可以动态创建、删除和修改 Topic 配置，这在多租户或自动化部署场景中非常实用。

@Service
public class TopicManager {

    @Autowired private KafkaAdmin kafkaAdmin;

    public void createTopicIfNotExists(String topicName, int partitions, short replicationFactor) {
        NewTopic newTopic = new NewTopic(topicName, partitions, replicationFactor);
        newTopic.configs(Map.of("retention.ms", String.valueOf(Duration.ofDays(7).toMillis())));
        kafkaAdmin.createOrModifyTopics(newTopic);
    }

    public void listTopics() {
        try (AdminClient client = AdminClient.create(kafkaAdmin.getConfigurationProperties())) {
            ListTopicsResult result = client.listTopics();
            result.names().get().forEach(System.out::println);
        }
    }
}

Spring Boot Actuator Kafka 健康检查

通过配置 management.health.kafka.enabled=true，Spring Boot Actuator 会自动检测 Kafka 连接状态，并将其纳入整体健康检查端点 /actuator/health。

management:
  endpoints:
    web:
      exposure:
        include: health,metrics,prometheus
  health:
    kafka:
      enabled: true

12. Schema Registry 与 Avro

Confluent Schema Registry 集成

在微服务系统中，不同服务之间共享消息格式经常面临向后兼容性问题。Confluent Schema Registry 提供了集中式的 Schema 管理，确保生产者发送的消息符合注册的模式定义，消费者也能按正确的 Schema 进行反序列化。

spring:
  kafka:
    producer:
      properties:
        schema.registry.url: http://schema-registry:8081
        value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
        auto.register.schemas: true
    consumer:
      properties:
        schema.registry.url: http://schema-registry:8081
        value.deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
        specific.avro.reader: true

Avro 序列化配置

首先需要定义 Avro Schema 文件 order.avsc：

{
  "type": "record",
  "name": "Order",
  "namespace": "com.example.model",
  "fields": [
    {"name": "id", "type": "string"},
    {"name": "userId", "type": "string"},
    {"name": "amount", "type": "double"},
    {"name": "status", "type": "string"}
  ]
}

使用 Avro Maven 插件生成 Java 类后，生产者可以直接发送 Avro 对象：

@Service
public class AvroOrderPublisher {
    private final KafkaTemplate<String, Order> kafkaTemplate;

    public void publishOrder(Order order) {
        kafkaTemplate.send("orders.avro", order.getUserId(), order);
    }
}

Schema Evolution 兼容性策略

Schema Registry 支持多种兼容性策略，控制 Schema 的变更规则：

BACKWARD：新 Schema 可以读取旧数据（新增可选字段、删除字段）
FORWARD：旧 Schema 可以读取新数据（新增可选字段）
FULL：同时满足 BACKWARD 和 FORWARD（只新增可选字段）
NONE：无任何兼容性检查

建议生产环境使用 BACKWARD 或 FULL 策略，避免因 Schema 变更导致消费者无法解析消息。在进行重大字段变更时，应先在测试环境验证兼容性，再更新生产环境的 Schema 注册。

13. 监控与可观测性

Spring Kafka 提供了丰富的可观测性接口，通过 KafkaListenerEndpointRegistry 可以获取所有消费者的运行指标，包括消费延迟、吞吐量、连接状态等。这些信息对于容量规划、故障排查和性能优化至关重要。

@Component
public class KafkaMetricsReporter {

    @Autowired private KafkaListenerEndpointRegistry registry;

    public void reportMetrics() {
        for (MessageListenerContainer container : registry.getListenerContainers()) {
            Map<String, Map<MetricName, ? extends Metric>> metrics = container.metrics();
            metrics.forEach((clientId, clientMetrics) -> {
                Metric recordsLag = clientMetrics.get(
                    new MetricName("records-lag-max", "consumer-fetch-manager-metrics", ""));
                if (recordsLag != null) {
                    log.info("Consumer {} max lag: {}", clientId, recordsLag.metricValue());
                }
            });
        }
    }
}

除了程序内获取指标，生产环境还应将日志输出到集中式日志系统（如 ELK 或 Loki），配合 Grafana 构建可视化监控面板，这样才能在系统出现异常时第一时间收到告警并定位根因。

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页