Spring-kafka消费者消费的一些问题

前言

Spring Kafka 无缝集成了 Spring Boot、Spring Framework 及其生态系统中的其他项目，如 Spring Cloud。通过与 Spring Boot 的自动配置结合，开发者可以快速启动和配置 Kafka 相关的功能。无需编写大量样板代码即可实现 Kafka 的生产和消费功能，在公司内部微服务项目中大量使用了该框架，以下是个人的一些记录。

笔者环境：springboot-2.7.14, java version “1.8.0_401”

一、Spring-kafka是什么？

Spring Kafka 是 Spring 框架的一个项目，旨在简化与 Apache Kafka 的集成。它提供了一组用于与 Kafka 通信的高级抽象和便利功能，使开发人员可以更轻松地在 Spring 应用程序中使用 Kafka 进行消息传递。

1. 什么是 Apache Kafka？

Apache Kafka 是一个分布式流处理平台，广泛用于构建实时数据管道和流应用程序。它具有以下几个关键功能：

发布和订阅记录流：类似于消息队列或企业消息传递系统。
存储记录流：存储数据流容错。
处理记录流：实时或批处理方式处理数据流。

2. 什么是 Spring Kafka？

Spring Kafka 是 Spring 框架的一个子项目，提供了用于与 Kafka 集成的便捷工具和抽象。Spring Kafka 包含在 Spring Boot 项目中，使得 Kafka 消息处理在 Spring 应用程序中变得更加容易和直观。

3. Spring Kafka 的主要功能

KafkaTemplate：用于发送消息的模板类。
@KafkaListener：用于消费消息的注解，支持自动并发和分区分配。
KafkaMessageListenerContainer：低级别的消息监听容器，提供了更大的灵活性。
事务支持：支持与 Kafka 的事务集成，确保消息的原子性和一致性。
错误处理：支持各种错误处理机制，包括重试、死信队列等。

4. 典型使用场景

消息驱动的微服务：使用 Kafka 作为微服务之间的消息传递中介。
实时数据处理：处理流式数据，如日志收集、监控和分析。
事件溯源：使用Kafka 记录所有事件，支持事件溯源和审计。
数据集成：连接不同的数据源和目标系统，实现数据集成和同步。

二、如何使用进行消息消费？

只需要简单四步步即可：引入依赖，配置消费者配置文件，配置 kafka configuration , 配置消费者监听器

1. 引入maven依赖

     <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

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

2. 配置文件

以下仅用于参考：

server:
  port: 8080
logging:
  level:
    root: info
spring:
  datasource:
    url: jdbc:mysql://localhost:3306/test
    username: root
    password: root
    driver-class-name: com.mysql.cj.jdbc.Driver
  # 开启事务管理
  transaction:
    management:
      enabled: true
  kafka:
    bootstrap-servers: localhost:9092
    # 消费监听接口监听的主题不存在时，默认会报错
    listener:
      missing-topics-fatal: false
    template:
      default-topic: events
    producer:
      # 当retris为0时，produce不会重复。retirs重发，此时repli节点完全成为leader节点，不会产生消息丢失。
      retries: 0
      #procedure要求leader在考虑完成请求之前收到的确认数，用于控制发送记录在服务端的持久化，其值可以为如下：
      #acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为-1。
      #acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
      #acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1的设置。
      #可以设置的值为：all, -1, 0, 1
      acks: 1
      # 指定消息key和消息体的编解码方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 每次批量发送消息的数量,produce积累到一定数据，一次发送
      #batch-size: 16384
      # produce积累数据一次发送，缓存大小达到buffer.memory就发送数据
      #buffer-memory: 33554432
    consumer:
      # 指定默认消费者group id --> 由于在kafka中，同一组中的consumer不会读取到同一个消息，依靠groud.id设置组名
      group-id: events-group-1
      # smallest和largest才有效，如果smallest重新0开始读取，如果是largest从logfile的offset读取。一般情况下我们都是设置smallest
      auto-offset-reset: earliest
      # enable.auto.commit:true --> 设置自动提交offset
      enable-auto-commit: true
      #如果'enable.auto.commit'为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000。
      auto-commit-interval: 100
      # 指定消息key和消息体的编解码方式
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      properties:
        spring.json.value.default.type: com.tan.kafka.model.Event


# 配置消费主题和消费者组2
kafka:
  topic: events_test
  group-2: events-group-2
  group-3: events-group-3
  group-5: events-group-5

里面的注释已经非常清晰了，相信都能看懂。

3. 配置kafka 消费者配置并使用注解启用


@EnableKafka
@Configuration
@Slf4j
public class KafkaConsumerConfig {

    /**
     * 创建一个Kafka消费者工厂，用于生产特定配置的Kafka消费者。
     * @param consumerFactory
     * @return
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Event> kafkaListenerContainerFactory(ConsumerFactory<String, Event> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, Event> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        log.info("KafkaListenerContainerFactory created.");
        return factory;
    }


    /**
     * 创建一个Kafka消费者工厂，用于生产特定配置的Kafka消费者。
     * 这个方法的主要作用是配置消费者的各项参数，包括连接服务器、消费者组ID、序列化器等，
     * 以便消费者能够正确地从Kafka主题中消费消息。
     *
     * @return ConsumerFactory<String, Event> 返回一个配置好的消费者工厂，用于创建字符串键和Event值的消费者实例。
     */
    @Bean
    public ConsumerFactory<String, Event> consumerFactory() {
        // 初始化配置属性映射，用于设置消费者的配置参数。
        Map<String, Object> configProps = new HashMap<>();
        // 配置Kafka服务器的连接地址和端口。
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 配置消费者的组ID，用于标识消费者属于哪个消费组。
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, "events-group");

        // 配置键和值的反序列化器，这里使用StringDeserializer和JsonDeserializer。
        // StringDeserializer用于反序列化键，JsonDeserializer用于反序列化值，并且信任所有包，设置默认值类型为Event类。
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        configProps.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
        configProps.put(JsonDeserializer.VALUE_DEFAULT_TYPE, Event.class);

        // 使用配置属性映射创建并返回一个默认的Kafka消费者工厂。
        return new DefaultKafkaConsumerFactory<>(configProps);
    }
}

⚠️ 这里的@EnableKafka 必不可少，因为它的主要作用是启用Kafka相关的注解，例如 @KafkaListener。当你在Spring Boot应用程序中使用 @EnableKafka 注解时，Spring会扫描你的项目，找到所有标注了 @KafkaListener 的方法，并创建相应的Kafka监听器容器来处理Kafka消息。

4. 配置监听器

/**
 * 同主题配置不同的消费组测试：每个消费者都可以获取到相同的消息
 * 同主题一个分区只能被多个相同消费组中的一个消费者消费
 */
@Service
public class EventConsumer {
    @Value("${kafka.topic}")
    public String topic;

    @Value("${kafka.group-2}")
    public String groupId;

    /**
     * 写法 1：指定具体消费者 topic,消费者组（这里消费者组可以覆盖消费者工厂中配置的消费者组）
     *
     * @param event
     */
    @KafkaListener(topics = "events", groupId = "events-group-1")
    public void consume(Event event) {
        System.out.println("Consumed event: " + event);
    }


    /**
     * 写法 2：使用 SpEL（Spring Expression Language）进行动态属性解析
     *  Spring 使用 SpEL 来解析注解中的表达式。在 Spring Kafka 中，
     *  __listener 提供了一种方式来引用当前监听器 bean 的上下文，从而可以动态地访问和配置监听器的属性。
     * 
     * @param message
     */
    @KafkaListener(topics = "#{__listener.topic}", groupId = "#{__listener.groupId}")
    public void consume2(Event message) {
        System.out.println("Consumed message: " + message);
    }


    /**
     * 写法 3：指定不同的消费者组
     *
     * @param event
     */
    @KafkaListener(topics = "${kafka.topic}", groupId = "${kafka.group-3}")
    public void consume3(Event event) {
        System.out.println("Consumer group-3 consumed event: " + event);
    }


    /**
     * 写法 4：
     * 1. 设置 ID 方便启动、和停止消费者
     * 2.可以指定 autoStartup 属性为 false，手动启动消费者,默认情况下是 true
     * @param event
     */
    @KafkaListener(id = "eventListener", topics = "events_test", groupId = "default", autoStartup = "true")
    public void consume5(Event event) {
        System.out.println("[event-listener] consumed event: " + event);
    }
}

注意：
__listener 特殊引用
__listener 是一个指向当前 Kafka 监听器的特殊 bean 引用。
通过使用 #{__listener.propertyName} 的方式，可以访问当前监听器 bean 的属性和方法。

以上不同的写法都可以，根据需要可以直接写消费者相关信息，也可以通过配置文件变量等方式获取，选择适合自己的即可。一般用配置变量的方式。

启动 springboot项目，然后发送消息，即可查看消息是否已经消费到了。

三、如何控制消费者启动和暂停？

上文中，我们可以在消费者的监听中加入autoStartup 来控制监听器是否在启动的时候开始进行消费监听，当然这种方式不够灵活，我们可以写一个接口在需要的时候进行启动和暂停，以下是个示例。

编写控制器


import com.tan.kafka.service.KafkaListenerControlService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


/**
 * @Description 控制消费者是否启动消费或者暂停
 * @Author zhengqiang.tan
 * @Date 2024/6/16 09:39
 */
@RestController
@RequestMapping("/kafka")
public class KafkaController {

    @Autowired
    private KafkaListenerControlService kafkaListenerControlService;

    @PostMapping("/pause")
    public String pause() {
        kafkaListenerControlService.pauseListener();
        return "Paused";
    }

    @PostMapping("/resume")
    public String resume() {
        kafkaListenerControlService.resumeListener();
        return "Resumed";
    }
}

编写服务类


/**
 * Kafka监听器控制服务
 */
@Service
@Slf4j
public class KafkaListenerControlService {

    @Autowired
    private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;
    
    @PostConstruct
    public void checkListeners() {
        log.info("KafkaListenerControlService.checkListeners");
        for (String listenerId : kafkaListenerEndpointRegistry.getListenerContainerIds()) {
            MessageListenerContainer container = kafkaListenerEndpointRegistry.getListenerContainer(listenerId);
            if (container != null) {
                System.out.println("Registered Kafka listener: " + listenerId);
            } else {
                System.out.println("No Kafka listener found with id: " + listenerId);
            }
        }
    }

    /**
     * 暂停指定名称的Kafka监听器。
     * <p>
     * 本方法旨在提供一种方式来暂停应用程序中特定的Kafka监听器。这可能在需要临时停止处理新消息，
     * 例如进行系统维护或升级时的场景下非常有用。
     * 这里使用 eventListener 仅用于测试。
     *
     * @see KafkaListenerEndpointRegistry#getListenerContainer(String) 用于获取指定名称的监听器容器
     * @see MessageListenerContainer#pause() 用于暂停监听器容器，使其停止接收和处理新消息
     */
    public void pauseListener() {
        // 通过监听器端点注册表获取名为"eventListener"的监听器容器
        MessageListenerContainer container = kafkaListenerEndpointRegistry.getListenerContainer("eventListener");
        // 检查容器是否为空，如果非空，则调用其pause方法暂停监听器
        if (container != null) {
            container.pause();
            log.info("Paused Kafka listener,id = {}", container.getListenerId());
        }
    }

    /**
     * 恢复指定名称的Kafka监听器。
     * 此方法用于暂停后的监听器重新启动，以便它可以继续处理消息。
     * 它首先从注册表中获取名为"eventListener"的监听器容器，
     * 然后检查容器是否为空。如果容器存在，则调用其resume方法来恢复监听器的运行。
     * 最后，记录一条信息表明监听器已恢复。
     *
     * @see KafkaListenerEndpointRegistry#getListenerContainer(String)
     * @see MessageListenerContainer#resume()
     */
    public void resumeListener() {
        // 从注册表中获取名为"eventListener"的监听器容器
        MessageListenerContainer container = kafkaListenerEndpointRegistry.getListenerContainer("eventListener");
        // 检查容器是否存在，如果存在，则恢复容器的运行
        if (container != null) {
            container.resume();
            // 记录恢复操作的信息，包括监听器的ID
            log.info("Resumed Kafka listener,id = {}", container.getListenerId());
        }
    }
}

通过以上方式，实测中符合预期，先暂停，然后发送消息后未进行消费，等到 resume之后继续消费。
如下所示：

curl --location --request POST 'http://localhost:8080/kafka/resume' \
--header 'User-Agent: Apifox/1.0.0 (https://apifox.com)' \
--header 'Content-Type: application/json' \
--header 'Accept: */*' \
--header 'Host: localhost:8080' \
--header 'Connection: keep-alive' \
--data-raw ''

在这里插入图片描述

四、如何保证消费者启停过程的线程安全问题？

如果pauseListener()方法被多个线程同时访问，可能会出现竞态条件或导致未定义的行为。尽管在这个特定的代码片段中不容易直接判断线程安全性问题。因此我在这里进行了完善，确保多线程并发访问下的安全性，以上写法太过繁琐，可以参考如下代码方式。

Slf4j
@RestController
@RequestMapping("/consumer")
public class KafkaController2 {
    private final ReentrantLock lock = new ReentrantLock();

    @Autowired
    private KafkaListenerEndpointRegistry registry;

    /**
     * 开启消费者监听，线程安全版本
     * @param listenerName
     */
    @GetMapping("/start/{listenerName}")
    public void start(@PathVariable String listenerName) {
        try {
            lock.lock();
            if (!isListenerRunning(listenerName)) {
                log.info("start {} ... ", listenerName);
                startListener(listenerName);
            }
            resumeListener(listenerName);
            log.info("resume {} is done. ", listenerName);
        } finally {
            lock.unlock();
        }
    }


    /**
     * 暂停指定名称的监听器。这有助于提高代码的灵活性和可维护性，
     * 允许在不修改代码的情况下暂停不同的监听器。
     *
     * @param listenerName 监听器的名称。
     */
    @GetMapping("/pause/{listenerName}")
    public void pauseListener(@PathVariable String listenerName) {
        if (registry == null) {
            log.error("Registry is not available.");
            return;
        }
        try {
            MessageListenerContainer listenerContainer = registry.getListenerContainer(listenerName);
            if (listenerContainer == null) {
                log.warn("No listener container found with name: " + listenerName);
                return;
            }
            log.info("Attempting to pause listener: " + listenerName);
            listenerContainer.pause();
            log.info(listenerName + " pause is done.");
        } catch (Exception e) {
            log.error("Failed to pause listener " + listenerName + ": " + e.getMessage(), e);
        }
    }


    @Async
    protected void startListener(String listenerName) {
        try {
            Objects.requireNonNull(registry.getListenerContainer(listenerName)).start();
        } catch (Exception e) {
            log.error("Failed to start the eventListener", e);
        }
    }

    @Async
    protected void resumeListener(String listenerName) {
        try {
            Objects.requireNonNull(registry.getListenerContainer(listenerName)).resume();
        } catch (Exception e) {
            log.error("Failed to resume the eventListener", e);
        }
    }

    private boolean isListenerRunning(String listenerName) {
        try {
            return Objects.requireNonNull(registry.getListenerContainer(listenerName)).isRunning();
        } catch (Exception e) {
            log.error("Error checking if the eventListener is running", e);
            return false;
        }
    }
}

最终，我在测试完后结果是符合预期的，先暂停，然后持续发送消息，观察有无收到消息，然后在重新启动监听即可。

curl --location --request GET 'http://localhost:8080/consumer/pause/eventListener'

在这里插入图片描述

其他

批量消息发送脚本，以下是个参考

#!/bin/bash
cd /Users/mac/apps/kafka
# 主题名称
TOPIC="events_test"

# Kafka 主机和端口
BROKER="localhost:9092"

# 生产随机消息的数量（只需修改这里进行控制发送消息数量）
NUM_MESSAGES=1

# 生成一个 5 位随机数
generate_random_number() {
    RANDOM_NUMBER=$(shuf -i 10000-99999 -n 1)
    echo $RANDOM_NUMBER
}

# 生成随机消息并发送到 Kafka 主题
for i in $(seq 1 $NUM_MESSAGES); do
    ID=$(date +%s) # 生成Long类型的唯一ID
    MSG_ID=$(uuidgen)
    MESSAGE="Random message $(generate_random_number)"
    JSON_MESSAGE="{\"id\":\"$ID\", \"msgId\":\"$MSG_ID\", \"message\":\"$MESSAGE\"}"
    echo $JSON_MESSAGE | bin/kafka-console-producer.sh --broker-list $BROKER --topic $TOPIC
done

echo "Produced $NUM_MESSAGES JSON messages to the topic '$TOPIC'."