Spring Boot整合Kafka实现消息消费的完整示例

爱谁谁

发布时间：2025-07-12 20:33:01

1001人浏览过

来源于php中文网

原创

spring boot整合kafka实现消息消费的核心在于简化配置和封装底层复杂性，使开发者专注于业务逻辑。1. 引入spring-kafka依赖；2. 配置kafka连接信息如服务器地址、消费者组、反序列化方式等；3. 使用@kafkalistener注解监听特定主题并处理消息，支持手动提交偏移量和批量消费；4. 自定义concurrentkafkalistenercontainerfactory以支持手动提交和批量消费场景。可靠性通过手动提交偏移量、错误处理机制（如死信队列）和合理配置消费者组参数保障；幂等性则依赖唯一业务id与持久化存储结合，确保重复消息不影响系统状态。配置陷阱包括group.id随意更改、auto-offset-reset误用、enable-auto-commit开启风险及max-poll-interval-ms与业务处理时间不协调。性能优化可通过调整concurrency参数提升并发度、启用批量消费减少提交开销、异步化耗时操作避免阻塞线程以及优化拉取参数降低资源消耗。重平衡应对策略包括合理设置session.timeout.ms与heartbeat.interval.ms、优雅停机、设计幂等消费者、监控消费者组状态及利用静态成员资格特性。总之，通过配置优化与健壮的消费者设计，可构建高可靠、高性能的kafka消息消费系统。

Spring Boot整合Kafka实现消息消费的完整示例

Spring Boot整合Kafka实现消息消费，核心在于它极大地简化了配置和编程模型，通过少量注解和配置就能快速搭建起一个健壮的消费者。它把底层Kafka客户端的复杂性封装得很好，让开发者可以更专注于业务逻辑本身。

解决方案

要用Spring Boot实现Kafka消息消费，通常你需要以下几个关键步骤和代码片段。

首先，在pom.xml里引入Spring Kafka的依赖：


    org.springframework.kafka
    spring-kafka

接着，在application.yml或application.properties中配置Kafka连接信息：

spring:
  kafka:
    bootstrap-servers: localhost:9092 # 你的Kafka服务器地址
    consumer:
      group-id: my-spring-consumer-group # 消费者组ID，非常重要
      auto-offset-reset: latest # 当没有初始偏移量或当前偏移量无效时，从哪里开始消费：earliest/latest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      enable-auto-commit: false # 推荐手动提交，以保证消息处理的可靠性
      max-poll-records: 500 # 每次poll操作最多获取的记录数
      fetch-min-bytes: 1 # 消费者从服务器获取数据时，最小等待字节数
      fetch-max-wait: 500 # 消费者从服务器获取数据时，最大等待时间（毫秒）

然后，创建一个消费者类，用@KafkaListener注解来监听特定主题：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class MyKafkaConsumer {

    // 监听名为 "my-topic" 的主题
    @KafkaListener(topics = "my-topic", groupId = "my-spring-consumer-group", containerFactory = "kafkaListenerContainerFactory")
    public void listen(String message) {
        System.out.println("收到消息 (字符串): " + message);
        // 如果是自动提交，这里就结束了
    }

    // 监听并手动提交偏移量，这在生产环境非常常见
    @KafkaListener(topics = "my-topic-manual-ack", groupId = "my-spring-consumer-group", containerFactory = "kafkaListenerContainerFactory")
    public void listenWithManualAck(ConsumerRecord record, Acknowledgment ack) {
        System.out.println("收到消息 (手动ack): Key=" + record.key() + ", Value=" + record.value() + ", Offset=" + record.offset());
        try {
            // 模拟业务处理
            Thread.sleep(100);
            if (record.value().contains("error")) {
                throw new RuntimeException("模拟处理错误");
            }
            // 业务处理成功后，手动提交偏移量
            ack.acknowledge();
            System.out.println("消息处理成功并提交偏移量: " + record.offset());
        } catch (Exception e) {
            System.err.println("消息处理失败，不提交偏移量: " + record.offset() + " 错误: " + e.getMessage());
            // 实际生产中，这里可能需要记录日志，或者将消息发送到死信队列 (DLQ)
            // 不调用 ack.acknowledge()，下次Kafka会重新投递这条消息
        }
    }

    // 批量消费，可以显著提高吞吐量
    @KafkaListener(topics = "my-topic-batch", groupId = "my-spring-consumer-group", containerFactory = "batchKafkaListenerContainerFactory")
    public void listenBatch(List> records, Acknowledgment ack) {
        System.out.println("收到批量消息，共 " + records.size() + " 条");
        for (ConsumerRecord record : records) {
            System.out.println("  - Key=" + record.key() + ", Value=" + record.value() + ", Offset=" + record.offset());
            // 批量处理逻辑
        }
        // 批量处理完成后，一次性提交整个批次的偏移量
        ack.acknowledge();
        System.out.println("批量消息处理完成并提交偏移量");
    }
}

最后，为了支持手动提交和批量消费，你可能需要自定义ConcurrentKafkaListenerContainerFactory。Spring Boot默认会提供一个，但我们可以覆盖它或者创建新的：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.beans.factory.annotation.Value;
import java.util.HashMap;
import java.util.Map;

@EnableKafka // 启用Kafka功能
@Configuration
public class KafkaConsumerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    @Value("${spring.kafka.consumer.auto-offset-reset}")
    private String autoOffsetReset;

    @Value("${spring.kafka.consumer.key-deserializer}")
    private String keyDeserializer;

    @Value("${spring.kafka.consumer.value-deserializer}")
    private String valueDeserializer;

    @Value("${spring.kafka.consumer.enable-auto-commit}")
    private Boolean enableAutoCommit;

    // 这是一个通用的消费者工厂配置，用于手动提交偏移量
    @Bean
    public ConsumerFactory consumerFactory() {
        Map props = new HashMap<>();
        props.put(org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        props.put(org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);
        props.put(org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
        props.put(org.apache.kafka.clients.consumer.ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
        // 如果要处理JSON，这里可以配置JsonDeserializer
        // props.put(JsonDeserializer.TRUSTED_PACKAGES, "*"); // 允许反序列化所有包下的类，生产环境慎用
        return new DefaultKafkaConsumerFactory<>(props);
    }

    // 普通的Kafka监听器容器工厂，用于单条消息消费
    @Bean
    public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        // 设置为手动提交偏移量
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        // 设置并发消费者数量，根据实际情况调整
        factory.setConcurrency(3); // 比如有3个分区，可以设置3个并发消费者
        return factory;
    }

    // 批量消费的Kafka监听器容器工厂
    @Bean
    public ConcurrentKafkaListenerContainerFactory batchKafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setBatchListener(true); // 启用批量监听
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        factory.setConcurrency(2);
        return factory;
    }
}

如何确保Kafka消息消费的可靠性和幂等性？

在我看来，消息消费的可靠性和幂等性是构建任何基于消息队列系统的基石，尤其在分布式环境下，这简直是绕不开的话题。可靠性确保消息不丢失，幂等性则保证重复处理同一条消息不会产生副作用。

可靠性方面：

首先，手动提交偏移量是确保可靠性的第一步，也是最关键的一步。Spring Kafka通过ContainerProperties.AckMode.MANUAL_IMMEDIATE或MANUAL_BATCH配合Acknowledgment.acknowledge()来支持。这意味着只有当你的业务逻辑真正处理完消息，并且确认无误后，才告诉Kafka可以安全地更新偏移量了。如果处理失败，不提交偏移量，那么这条消息会在下次消费时被重新拉取，这提供了一种“至少一次”的保证。我个人偏好MANUAL_IMMEDIATE，因为它能更及时地反映单条消息的处理状态。

其次，错误处理机制至关重要。如果消息处理失败，你不能就这么让它“消失”。Spring Kafka提供了ErrorHandler接口，你可以自定义一个实现来捕获监听器中的异常。比如，你可以将失败的消息发送到死信队列（Dead Letter Topic, DLT）。DLT是一个专门用来存放那些无法被正常处理的消息的主题，这样你就可以后续对这些消息进行人工干预、分析失败原因或者重试。这比简单地抛弃错误消息要负责得多。

再者，消费者组的配置也影响可靠性。session.timeout.ms和heartbeat.interval.ms决定了消费者多久没有心跳会被认为“挂了”从而触发重平衡。设置得太短，网络抖动可能导致不必要的重平衡；设置得太长，消费者真正挂掉后，Kafka需要更长时间才能发现并重新分配分区，这期间消息可能堆积。这是一个需要根据实际网络状况和业务容忍度去权衡的点。

幂等性方面：

DreamGen

一个AI驱动的角色扮演和故事写作的平台

下载

幂等性意味着对同一操作执行多次与执行一次的效果是相同的。在消息消费场景下，因为网络波动、消费者崩溃恢复、或者Kafka重试机制，消息很可能被重复投递。

实现幂等性通常需要业务层面的逻辑。最常见的方法是为每条消息设计一个唯一的业务ID。当消费者收到消息时，先检查这个ID是否已经被处理过。如果已经处理过，就直接丢弃（或者记录日志）；如果没有，则进行处理，并将这个ID标记为已处理。这个“已处理”的标记通常存储在数据库、Redis或者其他持久化存储中，并且这个检查和标记的过程需要是原子性的。

举个例子，如果你在处理订单支付消息，消息里通常会有一个transactionId。消费者在处理前，先去数据库里查一下这个transactionId对应的订单是否已经更新为“已支付”状态。如果是，就说明这条消息是重复的，直接跳过；如果不是，则进行支付状态更新，并将transactionId标记为已处理。这个操作本身需要是事务性的，确保数据一致性。

还有一种情况，如果你的业务操作本身就是幂等的，那就省心多了。比如，更新一个用户资料，无论你执行多少次UPDATE user SET name = 'new_name' WHERE id = 1，最终结果都是一样的。但这种场景并不总是适用。

Spring Boot Kafka消费者常见的配置陷阱与性能优化点有哪些？

在使用Spring Boot Kafka时，我踩过一些坑，也总结了一些优化经验。这些往往是配置不当或者对Kafka内部机制理解不足导致的。

常见的配置陷阱：

group.id 的随意更改：这是个大坑。每次启动应用，如果group.id变了，Kafka会认为这是一个全新的消费者组，从头开始消费（取决于auto-offset-reset），这会导致消息重复消费，尤其在生产环境，后果很严重。所以，group.id一旦确定，就不要轻易改动，除非你确实需要一个新的消费逻辑。
auto-offset-reset 的误用：earliest和latest的选择非常重要。earliest会从最早的可用偏移量开始消费，可能导致大量历史消息被重复处理；latest则从最新的消息开始。开发测试时用earliest方便看所有消息，但生产环境通常用latest，或者在首次启动时用earliest，之后改为latest。
enable-auto-commit 为 true 时的风险：虽然方便，但如果消息处理失败了，而偏移量已经自动提交了，那这条失败的消息就“丢”了，不会再被重新处理。所以，我强烈建议在生产环境设置为false并手动提交。
max-poll-records 和 max-poll-interval.ms 的不协调：max-poll-records控制单次拉取的消息数量，max-poll-interval.ms是消费者在两次poll操作之间允许的最长时间。如果你的业务处理逻辑很耗时，导致处理一批消息的时间超过了max-poll-interval.ms，那么Kafka会认为这个消费者“死了”，从而触发重平衡，将分区分配给其他消费者。这会导致消息重复处理。你需要确保你的处理速度足够快，或者适当调大max-poll-interval.ms，但这又可能延长检测消费者失败的时间。
反序列化失败：这是个很常见的问题。如果Kafka消息的格式和你消费者期望的不一致（比如JSON结构变了，或者发了一个非JSON字符串），value-deserializer就会抛异常。Spring Kafka默认会停止整个容器。你可以通过配置ErrorHandlingDeserializer并结合一个FailedDeserializationFunction来处理这些反序列化错误，比如把错误消息记录下来或者发送到DLT，而不是让整个消费者停掉。

性能优化点：

concurrency 参数：这是ConcurrentKafkaListenerContainerFactory的关键。它决定了有多少个线程同时去消费分区。如果你有多个分区，设置合适的concurrency（通常不超过分区数）可以显著提高吞吐量。比如，一个主题有3个分区，你可以设置concurrency为3，这样每个分区都有一个线程在独立消费。
批量消费 (@KafkaListener(batch = true))：如果你需要处理大量小消息，批量消费能大幅减少网络往返和提交偏移量的开销。一次性拉取多条消息，在业务逻辑中批量处理，然后一次性提交偏移量。这对于数据库批处理插入等场景尤其有效。
优化消息处理逻辑：无论Kafka配置如何，最终性能瓶颈往往在你的业务处理代码上。如果你的监听器方法里有IO操作（如数据库查询/写入、调用外部服务），考虑将这些操作异步化，或者使用线程池来处理，避免阻塞Kafka消费线程。比如，你可以将消息放入一个内部队列，然后由另一个线程池异步处理。
调整 fetch.min.bytes 和 fetch.max.wait.ms：这两个参数影响消费者拉取消息的行为。fetch.min.bytes设置了Kafka服务器在返回数据前，需要累积的最小字节数。fetch.max.wait.ms是等待达到fetch.min.bytes的最长时间。适当调大它们，可以在消息量不大时减少poll请求的频率，降低CPU和网络开销，但会增加一点点延迟。
合理使用分区：主题的分区数量直接影响并发度。分区越多，理论上可以支持的消费者并发度越高。但分区也不是越多越好，过多的分区会增加Kafka集群的元数据管理开销和消费者重平衡时的成本。

面对Kafka消费者组重平衡，我们应该如何应对？

消费者组重平衡，这事儿在Kafka的世界里是再正常不过的了，但处理不好确实会让人头疼。简单来说，重平衡就是当消费者组的成员发生变化（比如有新的消费者加入、现有消费者离开或崩溃），或者主题的分区数量发生变化时，Kafka会重新分配分区给组内的所有活跃消费者，以确保每个分区都被且仅被一个消费者消费。

重平衡的影响：

最直接的影响就是消息消费的暂停。在重平衡期间，消费者会停止拉取消息，直到新的分区分配方案确定并生效。这会导致消费延迟。另一个潜在问题是消息重复处理。如果一个消费者在处理完消息但未来得及提交偏移量时崩溃，那么在重平衡后，这个分区可能会被分配给另一个消费者，而新的消费者会从上一个已提交的偏移量开始拉取，从而可能重复处理崩溃消费者未提交的那部分消息。这就是为什么我一直强调幂等性设计的重要性。

应对策略：

理解和配置 session.timeout.ms 和 heartbeat.interval.ms：
- session.timeout.ms：这是消费者被认为“活跃”的最长时间，如果在此时间内没有发送心跳，它就会被认为失败并从组中移除，触发重平衡。
- heartbeat.interval.ms：消费者发送心跳的频率。通常建议设置为session.timeout.ms的1/3左右。合理配置这两个参数是减少不必要重平衡的关键。如果你的消费者处理逻辑比较慢，导致无法及时发送心跳，可以适当调大session.timeout.ms，但不要过大，否则消费者真正挂掉后，Kafka需要更长时间才能发现。
优雅停机：当你的Spring Boot应用关闭时，要确保Kafka消费者能够优雅地停止，释放持有的分区。Spring Kafka的ConcurrentMessageListenerContainer默认会处理这个，但你也可以通过配置container.properties.setShutdownTimeout()来设置一个超时时间，让消费者在关闭前有足够的时间完成当前正在处理的消息并提交偏移量。这能有效避免因突然关闭而触发的重平衡。
设计幂等性消费者：这是应对重平衡导致消息重复的根本之道。无论重平衡如何发生，只要你的业务逻辑是幂等的，重复处理同一条消息就不会产生副作用。我已经提过，通过唯一的业务ID来判断消息是否已处理是常见做法。
监控消费者组状态：使用Kafka提供的工具（如kafka-consumer-groups.sh）或者集成到你的监控系统（如Prometheus + Grafana）中，监控消费者组的成员变化、分区分配情况以及消费者延迟（consumer lag）。及时发现并定位重平衡发生的原因，是解决问题的起点。
利用Kafka 2.3+ 的静态成员资格（Static Membership）：这是一个相对较新的特性。通过为消费者设置一个group.instance.id，即使消费者重启，只要它的group.instance.id不变，Kafka就会认为它是同一个实例，而不会触发重平衡，直到它真正停止运行或者ID改变。这大大减少了因为消费者重启而引起的重平衡次数，提高了系统的可用性和稳定性。当然，这需要你的Kafka集群版本支持。
避免在监听器内部执行长时间阻塞操作：如果你的@KafkaListener方法内部有长时间的阻塞操作，可能会导致消费者无法及时发送心跳，从而被踢出消费者组，引发重平衡。如果确实需要执行耗时操作，考虑将消息异步投递到另一个线程池中处理，让Kafka消费线程尽快返回，保持活跃。