当前位置：首页 > news >正文

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

news 文章来源：https://blog.csdn.net/u010834071/article/details/139720372 2024/10/6 6:27:37

前言

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/testusername: rootpassword: rootdriver-class-name: com.mysql.cj.jdbc.Driver# 开启事务管理transaction:management:enabled: truekafka:bootstrap-servers: localhost:9092# 消费监听接口监听的主题不存在时，默认会报错listener:missing-topics-fatal: falsetemplate:default-topic: eventsproducer:# 当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, 1acks: 1# 指定消息key和消息体的编解码方式key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializer# 每次批量发送消息的数量,produce积累到一定数据，一次发送#batch-size: 16384# produce积累数据一次发送，缓存大小达到buffer.memory就发送数据#buffer-memory: 33554432consumer:# 指定默认消费者group id --> 由于在kafka中，同一组中的consumer不会读取到同一个消息，依靠groud.id设置组名group-id: events-group-1# smallest和largest才有效，如果smallest重新0开始读取，如果是largest从logfile的offset读取。一般情况下我们都是设置smallestauto-offset-reset: earliest# enable.auto.commit:true --> 设置自动提交offsetenable-auto-commit: true#如果'enable.auto.commit'为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000。auto-commit-interval: 100# 指定消息key和消息体的编解码方式key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializerproperties:spring.json.value.default.type: com.tan.kafka.model.Event# 配置消费主题和消费者组2
kafka:topic: events_testgroup-2: events-group-2group-3: events-group-3group-5: events-group-5

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

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


@EnableKafka
@Configuration
@Slf4j
public class KafkaConsumerConfig {/*** 创建一个Kafka消费者工厂，用于生产特定配置的Kafka消费者。* @param consumerFactory* @return*/@Beanpublic 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值的消费者实例。*/@Beanpublic 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 {@Autowiredprivate 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 {@Autowiredprivate KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;@PostConstructpublic 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();// 记录恢复操作的信息，包括监听器的IDlog.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();@Autowiredprivate 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);}}@Asyncprotected void startListener(String listenerName) {try {Objects.requireNonNull(registry.getListenerContainer(listenerName)).start();} catch (Exception e) {log.error("Failed to start the eventListener", e);}}@Asyncprotected 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'