0
0

【系统设计】提升Kafka系统性能：Spring Boot实现Lag感知的生产者与消费者

在分布式系统中，消息队列扮演着至关重要的角色，而Kafka凭借其高吞吐量和低延迟的特性，成为了流数据处理的首选。然而，随着应用规模的扩大，如何有效管理Kafka的offset与lag，确保系统的高可用性和性能稳定，成为了一个亟待解决的问题。本文将深入探讨如何实现一个基于Key的Lag感知生产者与消费者，通过动态监控与调节，实现智能的负载均衡，从而在部分服务器性能下降时，依然能够保证整体系统的高效

Kafka Offset与Lag的概念解析

在Kafka中，Offset和Lag是两个关键的指标，用于监控和管理消费者的消费进度。

Offset（偏移量）

每个Kafka分区中的消息都有一个唯一的offset，用于标识该消息在分区中的位置。消费者在消费消息时，会记录其最新消费的offset，以便在出现故障或重启时能够从上次消费的位置继续。

Lag（延迟）

Lag表示消费者与生产者之间的延迟，即生产者生成的最新消息的offset与消费者已消费的offset之间的差值。Lag越大，意味着消费者处理消息的速度跟不上生产者的速度，可能导致系统的延迟增加甚至出现消息堆积。

Offset与Lag的计算方式

  • Offset计算：每个消费者组在每个分区上都会维护一个offset，表示该消费者组在该分区上已经消费到的位置。
  • Lag计算：Lag = 生产者最新的offset - 消费者当前的offset

通过定期监控每个分区的Lag值，可以及时发现消费滞后的问题，并采取相应的措施，如扩展消费者实例或调整生产者的负载策略。

消费者的工作机制及负载均衡实现

消费者的工作机制

Kafka消费者通过订阅一个或多个主题（Topic）来消费消息。消费者组（Consumer Group）允许多个消费者实例共同消费一个主题的消息，每个分区只能被一个消费者实例消费，从而实现负载均衡。

主要步骤：

  1. 订阅主题：消费者订阅一个或多个主题。
  2. 分配分区：Kafka会将主题的分区分配给消费者组中的各个消费者实例，确保每个分区只被一个消费者消费。
  3. 消费消息：消费者按照分配的分区顺序消费消息，并定期提交offset。
  4. 负载均衡：当消费者数量变化（增加或减少）时，Kafka会自动重新分配分区，确保负载均衡。

负载均衡的实现

在传统的负载均衡中，分区的分配是静态的，无法根据实时的Lag情况进行动态调整。为了实现基于Lag的动态负载均衡，可以采取以下策略：

  1. 监控Lag：定期监控每个分区的Lag值，识别出Lag较高的分区。
  2. 动态调整分区分配： - 当某个分区的Lag超过预设阈值时，消费者主动退出对该分区的订阅。- 触发消费者组的再平衡机制，使其他消费者接手高Lag分区的消费任务。
  3. 重新分配消费者：通过消费者组的再平衡机制，使其他消费者实例接手高Lag分区的负载，实现负载均衡。

这种动态调整机制能够有效应对部分服务器性能下降的情况，提升整体系统的稳定性和性能。

基于Spring Boot的实现源码

下面将通过一个Spring Boot项目，详细展示如何实现基于Key的Lag感知生产者与消费者，以及动态负载均衡的具体实现。

项目结构

kafka-lag-aware
├── src
│   ├── main
│   │   ├── java
│   │   │   └── com.example.kafkalagaware
│   │   │       ├── KafkaLagAwareApplication.java
│   │   │       ├── config
│   │   │       │   ├── KafkaConsumerConfig.java
│   │   │       │   └── KafkaProducerConfig.java
│   │   │       ├── consumer
│   │   │       │   └── LagAwareConsumer.java
│   │   │       ├── partitioner
│   │   │       │   └── LagAwarePartitioner.java
│   │   │       └── producer
│   │   │           └── LagAwareProducer.java
│   │   └── resources
│   │       └── application.yml
└── pom.xml

依赖配置

在

pom.xml

中添加必要的依赖：

<projectxmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="
        http://maven.apache.org/POM/4.0.0 
        http://maven.apache.org/maven-v4_0_0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.example</groupId><artifactId>kafka-lag-aware</artifactId><version>1.0.0</version><packaging>jar</packaging><name>kafka-lag-aware</name><description>Kafka Lag Aware Producer and Consumer with Dynamic Load Balancing</description><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>3.1.2</version><relativePath/><!-- lookup parent from repository --></parent><dependencies><!-- Spring Boot Starter --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><!-- Spring for Apache Kafka --><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency><!-- Lombok for boilerplate code reduction --><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><!-- Additional dependencies if needed --></dependencies><build><plugins><!-- Spring Boot Maven Plugin --><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin><!-- Lombok Plugin --><plugin><groupId>org.projectlombok</groupId><artifactId>lombok-maven-plugin</artifactId><version>1.18.24.0</version><executions><execution><phase>compile</phase><goals><goal>delombok</goal></goals></execution></executions></plugin></plugins></build></project>

配置文件

application.yml

spring:kafka:bootstrap-servers: localhost:9092consumer:group-id: lag-aware-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      enable-auto-commit:falseauto-offset-reset: earliest
    producer:key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      partitioner: com.example.kafkalagaware.partitioner.LagAwarePartitioner
      properties:bootstrap.servers: localhost:9092listener:type: batch
    topic: my-topic
    lag:threshold:100

Kafka配置类

a.
KafkaProducerConfig.java

确保在生产者配置中正确设置自定义分区器。

packagecom.example.kafkalagaware.config;importorg.apache.kafka.clients.producer.ProducerConfig;importorg.apache.kafka.common.serialization.StringSerializer;importorg.springframework.beans.factory.annotation.Value;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importorg.springframework.kafka.core.*;importjava.util.HashMap;importjava.util.Map;@ConfigurationpublicclassKafkaProducerConfig{@Value("${spring.kafka.bootstrap-servers}")privateString bootstrapServers;@BeanpublicProducerFactory<String,String>producerFactory(){Map<String,Object> props =newHashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class);// 设置自定义分区器
        props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.example.kafkalagaware.partitioner.LagAwarePartitioner");// Additional producer configurations if neededreturnnewDefaultKafkaProducerFactory<>(props);}@BeanpublicKafkaTemplate<String,String>kafkaTemplate(){returnnewKafkaTemplate<>(producerFactory());}}
b.
KafkaConsumerConfig.java

packagecom.example.kafkalagaware.config;importorg.apache.kafka.clients.consumer.ConsumerConfig;importorg.apache.kafka.common.serialization.StringDeserializer;importorg.springframework.beans.factory.annotation.Value;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importorg.springframework.kafka.core.*;importorg.springframework.kafka.listener.*;importjava.util.HashMap;importjava.util.Map;@ConfigurationpublicclassKafkaConsumerConfig{@Value("${spring.kafka.bootstrap-servers}")privateString bootstrapServers;@Value("${spring.kafka.consumer.group-id}")privateString groupId;@BeanpublicConsumerFactory<String,String>consumerFactory(){Map<String,Object> props =newHashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");// Additional consumer configurations if neededreturnnewDefaultKafkaConsumerFactory<>(props);}@BeanpublicConcurrentKafkaListenerContainerFactory<String,String>kafkaListenerContainerFactory(){ConcurrentKafkaListenerContainerFactory<String,String> factory =newConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);// Number of consumer threads
        factory.setBatchListener(true);return factory;}}

实现基于Key的Lag感知生产者

LagAwareProducer 负责发送消息。由于我们已经在自定义分区器中处理了分区选择逻辑，因此 LagAwareProducer 只需负责发送消息即可，无需手动选择分区。

packagecom.example.kafkalagaware.producer;importorg.springframework.beans.factory.annotation.*;importorg.springframework.kafka.core.KafkaTemplate;importorg.springframework.stereotype.Component;@ComponentpublicclassLagAwareProducer{@AutowiredprivateKafkaTemplate<String,String> kafkaTemplate;@Value("${spring.kafka.topic}")privateString topic;/**
     * 发送消息，依赖自定义分区器决定目标分区
     */publicvoidsendMessage(String key,String value){
        kafkaTemplate.send(topic, key, value).addCallback(
                success ->System.out.println("Message sent to partition "+ success.getRecordMetadata().partition()),
                failure ->System.err.println("Failed to send message: "+ failure.getMessage()));}}

实现Lag感知的消费者

LagAwareConsumer 负责消费消息，并监控自身的 Lag 值。当 Lag 超过阈值时，消费者主动退出对该分区的订阅，触发消费者组的再平衡机制，使其他消费者接管该分区。

packagecom.example.kafkalagaware.consumer;importorg.apache.kafka.clients.consumer.Consumer;importorg.apache.kafka.clients.consumer.ConsumerRecord;importorg.apache.kafka.common.TopicPartition;importorg.springframework.beans.factory.annotation.*;importorg.springframework.kafka.annotation.KafkaListener;importorg.springframework.kafka.core.ConsumerFactory;importorg.springframework.kafka.support.Acknowledgment;importorg.springframework.scheduling.annotation.Scheduled;importorg.springframework.stereotype.Component;importjava.util.*;importjava.util.concurrent.ConcurrentHashMap;@ComponentpublicclassLagAwareConsumer{@Value("${spring.kafka.lag.threshold:100}")privatelong lagThreshold;@AutowiredprivateConsumerFactory<String,String> consumerFactory;privatefinalSet<TopicPartition> pausedPartitions =ConcurrentHashMap.newKeySet();@KafkaListener(topics ="${spring.kafka.topic}", containerFactory ="kafkaListenerContainerFactory")publicvoidlisten(List<ConsumerRecord<String,String>> records,Acknowledgment acknowledgment,Consumer<?,?> consumer){for(ConsumerRecord<String,String> record : records){// 处理消息System.out.printf("Consumed message: key=%s, value=%s, partition=%d, offset=%d%n",
                    record.key(), record.value(), record.partition(), record.offset());// 检查Laglong lag =getLag(record.topic(), record.partition(), consumer);if(lag > lagThreshold){TopicPartition partition =newTopicPartition(record.topic(), record.partition());if(!pausedPartitions.contains(partition)){
                    consumer.pause(Collections.singleton(partition));
                    pausedPartitions.add(partition);System.out.println("Paused partition: "+ partition.partition());// 主动取消订阅该分区，触发再平衡unsubscribeFromPartition(consumer, partition);}}}
        acknowledgment.acknowledge();}/**
     * 获取分区的Lag值
     */privatelonggetLag(String topic,int partition,Consumer<?,?> consumer){TopicPartition topicPartition =newTopicPartition(topic, partition);long latestOffset = consumer.endOffsets(Collections.singleton(topicPartition)).get(topicPartition);long currentOffset = consumer.position(topicPartition);return latestOffset - currentOffset;}/**
     * 主动取消订阅高Lag的分区，触发消费者组的再平衡
     */privatevoidunsubscribeFromPartition(Consumer<?,?> consumer,TopicPartition partition){try{// 获取当前分配的分区Set<TopicPartition> currentPartitions = consumer.assignment();// 移除高Lag分区Set<TopicPartition> updatedPartitions =newHashSet<>(currentPartitions);
            updatedPartitions.remove(partition);// 重新分配分区
            consumer.assign(updatedPartitions);System.out.println("Unsubscribed from partition: "+ partition.partition());}catch(Exception e){
            e.printStackTrace();}}/**
     * 定期检查暂停的分区是否可以恢复消费
     */@Scheduled(fixedDelay =10000)publicvoidcheckPausedPartitions(){// 此处需要实现逻辑检查Lag是否恢复// 如果Lag已经降低，可以移除pausedPartitions中的分区，并重新订阅// 具体实现请根据实际需求编写}}

实现自定义分区器

LagAwarePartitioner 实现自定义分区选择逻辑，根据每个分区的 Lag 值，选择 Lag 最小的分区来发送消息，确保负载均衡。

packagecom.example.kafkalagaware.partitioner;importorg.apache.kafka.clients.admin.*;importorg.apache.kafka.clients.producer.Partitioner;importorg.apache.kafka.common.Cluster;importorg.apache.kafka.common.PartitionInfo;importorg.apache.kafka.common.TopicPartition;importorg.apache.kafka.common.utils.Utils;importjava.util.*;importjava.util.concurrent.ExecutionException;importjava.util.stream.Collectors;publicclassLagAwarePartitionerimplementsPartitioner{privateAdminClient adminClient;privateString consumerGroupId ="lag-aware-group";// 消费者组ID@Overridepublicvoidconfigure(Map<String,?> configs){Properties props =newProperties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, configs.get("bootstrap.servers"));this.adminClient =AdminClient.create(props);}@Overridepublicintpartition(String topic,Object keyObj,byte[] keyBytes,Object value,byte[] valueBytes,Cluster cluster){String key =(String) keyObj;List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);List<Integer> partitionNumbers = partitions.stream().map(PartitionInfo::partition).collect(Collectors.toList());try{// 获取每个分区的Lag值Map<TopicPartition,Long> lagMap =getLagMap(topic, partitionNumbers);int totalPartitions = partitions.size();// 根据Key的哈希值选择默认分区int hash =Math.abs(key.hashCode());int defaultPartition = hash % totalPartitions;TopicPartition defaultTp =newTopicPartition(topic, defaultPartition);long defaultLag = lagMap.getOrDefault(defaultTp,0L);if(defaultLag >=100){// 预设的Lag阈值// 如果默认分区Lag过高，则选择Lag最小的分区Optional<TopicPartition> optionalPartition = lagMap.entrySet().stream().filter(entry -> entry.getValue()<100).min(Comparator.comparingLong(Map.Entry::getValue)).map(Map.Entry::getKey);if(optionalPartition.isPresent()){return optionalPartition.get().partition();}}// 默认分区Lag正常，发送到默认分区return defaultPartition;}catch(Exception e){
            e.printStackTrace();// 回退到默认的轮询策略returndefaultPartition(keyBytes, partitions, cluster);}}privateintdefaultPartition(byte[] keyBytes,List<PartitionInfo> partitions,Cluster cluster){if(keyBytes ==null){// 如果没有key，则随机选择分区returnnewRandom().nextInt(partitions.size());}else{// 使用默认的hash策略returnUtils.toPositive(Utils.murmur2(keyBytes))% partitions.size();}}privateMap<TopicPartition,Long>getLagMap(String topic,List<Integer> partitions)throwsExecutionException,InterruptedException{Map<TopicPartition,Long> lagMap =newHashMap<>();List<TopicPartition> topicPartitions = partitions.stream().map(p ->newTopicPartition(topic, p)).collect(Collectors.toList());// 获取最新的offsetListOffsetsResult latestResult = adminClient.listOffsets(
                topicPartitions.stream().collect(Collectors.toMap(tp -> tp, tp ->OffsetSpec.latest())));Map<TopicPartition,ListOffsetsResult.ListOffsetsResultInfo> latestOffsets = latestResult.all().get();// 获取消费者组的当前offsetListConsumerGroupOffsetsResult consumerOffsetsResult = adminClient.listConsumerGroupOffsets(consumerGroupId);Map<TopicPartition,OffsetAndMetadata> consumerOffsets = consumerOffsetsResult.partitionsToOffsetAndMetadata().get();for(TopicPartition tp : topicPartitions){long latestOffset = latestOffsets.get(tp).offset();long consumerOffset = consumerOffsets.getOrDefault(tp,newOffsetAndMetadata(0L)).offset();long lag = latestOffset - consumerOffset;
            lagMap.put(tp, lag);}return lagMap;}@Overridepublicvoidclose(){if(adminClient !=null){
            adminClient.close();}}@OverridepublicvoidonNewBatch(String topic,Cluster cluster,int prevPartition){// 可选实现}}

注意事项：

  • 性能开销：自定义分区器中调用AdminClient获取Lag信息会带来额外的网络和计算开销，特别是在高频率发送消息的场景下。因此，建议对Lag信息进行缓存，并设置合理的更新频率。
  • 消息顺序性：使用Key进行分区选择时，确保具有相同Key的消息发送到同一分区，以保持顺序性。在本实现中，当默认分区Lag过高时，可能会将相同Key的消息发送到不同分区，从而打乱顺序。根据需求，可能需要调整策略以平衡负载均衡与顺序性。
  • 故障处理：在自定义分区器中，需要妥善处理AdminClient的异常情况，确保在无法获取Lag信息时，系统能够退化到默认的分区策略，避免消息发送失败。

主应用类

KafkaLagAwareApplication

作为Spring Boot的入口，初始化Producer和消费者。

packagecom.example.kafkalagaware;importcom.example.kafkalagaware.producer.LagAwareProducer;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.boot.*;importorg.springframework.boot.autoconfigure.SpringBootApplication;importorg.springframework.scheduling.annotation.EnableScheduling;@SpringBootApplication@EnableSchedulingpublicclassKafkaLagAwareApplicationimplementsCommandLineRunner{@AutowiredprivateLagAwareProducer producer;publicstaticvoidmain(String[] args){SpringApplication.run(KafkaLagAwareApplication.class, args);}@Overridepublicvoidrun(String... args)throwsException{// 模拟生产消息for(int i =0; i <1000; i++){String key ="key-"+(i %50);// 使用有限的Key，增加分区的重复使用String value ="value-"+ i;
            producer.sendMessage(key, value);Thread.sleep(10);// 控制生产速率}}}

总结

本文详细介绍了如何在Kafka中实现一个基于Key的Lag感知生产者与消费者，通过监控各分区的Lag值，动态调整消息的生产与消费策略，达到智能的负载均衡效果。具体实现中：

  1. 生产者：通过自定义分区器 LagAwarePartitioner，在发送消息时根据 Key 的哈希值和分区的 Lag 值选择目标分区。当默认分区的 Lag 超过预设阈值时，自动选择 Lag 最小的分区发送消息。
  2. 消费者：在消费过程中持续监控自身的 Lag 值，当 Lag 超过预设阈值时，主动退出对该分区的订阅，触发消费者组的再平衡机制，使其他消费者接管该分区。
  3. 动态负载均衡：通过生产者和消费者的协同工作，确保高 Lag 分区能够被性能较好的消费者接管，提升整体系统的稳定性和性能。

应用场景：这种机制特别适用于服务器性能不均衡的情况下，能够显著提升整体系统的稳定性和性能，避免因部分节点性能问题导致整个系统的瓶颈。

通过Spring Boot与Spring Kafka的结合，简化了配置与开发流程，使得实现复杂的Kafka消费者逻辑变得更加高效和便捷。未来，可以进一步优化Lag检测与负载转移的策略，例如引入更多的监控指标，结合自动伸缩机制，实现更加智能化的系统自我调节能力。同时，结合容器化和微服务架构，能够在更大规模的分布式环境中，充分发挥基于Key的Lag感知机制的优势，保障数据流处理的高效与可靠。

标签： spring boot kafka 架构
本文转载自: https://blog.csdn.net/yhkal/article/details/143467252
版权归原作者 0_1_bits 所有， 如有侵权，请联系我们删除。
登录后发布评论

“【系统设计】提升Kafka系统性能：Spring Boot实现Lag感知的生产者与消费者”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航