kafka消息堆积及分区不均匀的解决方案

墨蓝 2023-07-17 12:38 2阅读 0赞

我在环境中发现代码里面的kafka有所延迟，查看kafka消息发现堆积严重，经过检查发现是kafka消息分区不均匀造成的，消费速度过慢。这里由自己在虚拟机上演示相关问题，给大家提供相应问题的参考思路。  
这篇文章有点遗憾并没重现分区不均衡的样例和Warning: Consumer group ‘testGroup1’ is rebalancing. 这里仅将正确的方式展示，等后续重现了在进行补充。

主要有两个要点：

1、一个消费者组只消费一个topic.  
2、factory.setConcurrency(concurrency)；这里设置监听并发数为 部署单元节点\*concurrency=分区数量

1、先在kafka消息中创建对应分区数目的topic（testTopic2，testTopic3）testTopic1由代码创建

./kafka-topics.sh --create --zookeeper 192.168.25.128:2181 --replication-factor 1 --partitions 2 --topic testTopic2

2、添加配置文件application.properties

kafka.test.topic1=testTopic1
    kafka.test.topic2=testTopic2
    kafka.test.topic3=testTopic3
    kafka.broker=192.168.25.128:9092
    auto.commit.interval.time=60000
    #kafka.test.group=customer-test
    kafka.test.group1=testGroup1
    kafka.test.group2=testGroup2
    kafka.test.group3=testGroup3
    kafka.offset=earliest
    kafka.auto.commit=false
    
    session.timeout.time=10000
    kafka.concurrency=2

3、创建kafka工厂

package com.yin.customer.config;
    
    import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.common.serialization.StringDeserializer;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
    import org.springframework.kafka.config.KafkaListenerContainerFactory;
    import org.springframework.kafka.core.ConsumerFactory;
    import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
    import org.springframework.kafka.listener.AbstractMessageListenerContainer;
    import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
    import org.springframework.kafka.listener.ContainerProperties;
    import org.springframework.stereotype.Component;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * @author yin
     * @Date 2019/11/24 15:54
     * @Method
     */
    @Configuration
    @Component
    public class KafkaConfig {
        @Value("${kafka.broker}")
        private String broker;
        @Value("${kafka.auto.commit}")
        private String autoCommit;
    
       // @Value("${kafka.test.group}")
        //private String testGroup;
    
        @Value("${session.timeout.time}")
        private String sessionOutTime;
    
        @Value("${auto.commit.interval.time}")
        private String autoCommitTime;
    
        @Value("${kafka.offset}")
        private String offset;
        @Value("${kafka.concurrency}")
        private Integer concurrency;
    
    
       @Bean
        KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory(){
            ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactory());
            //监听设置两个个分区
            factory.setConcurrency(concurrency);
            //打开批量拉取数据
            factory.setBatchListener(true);
            //这里设置的是心跳时间也是拉的时间，也就说每间隔max.poll.interval.ms我们就调用一次poll,kafka默认是300s,心跳只能在poll的时候发出，如果连续两次poll的时候超过
            //max.poll.interval.ms 值就会导致rebalance
            //心跳导致GroupCoordinator以为本地consumer节点挂掉了，引发了partition在consumerGroup里的rebalance。
            // 当rebalance后，之前该consumer拥有的分区和offset信息就失效了，同时导致不断的报auto offset commit failed。
            factory.getContainerProperties().setPollTimeout(3000);
            factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
            return factory;
        }
    
        private ConsumerFactory<String,String> consumerFactory() {
            return new DefaultKafkaConsumerFactory<String, String>(consumerConfigs());
        }
    
    
    
       @Bean
        public Map<String, Object> consumerConfigs() {
            Map<String, Object> propsMap = new HashMap<>();
            //kafka的地址
            propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, broker);
            //是否自动提交 Offset
            propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, autoCommit);
            // enable.auto.commit 设置成 false，那么 auto.commit.interval.ms 也就不被再考虑
            //默认5秒钟，一个 Consumer 将会提交它的 Offset 给 Kafka
            propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,  5000);
    
            //这个值必须设置在broker configuration中的group.min.session.timeout.ms 与 group.max.session.timeout.ms之间。
            //zookeeper.session.timeout.ms 默认值：6000
            //ZooKeeper的session的超时时间，如果在这段时间内没有收到ZK的心跳，则会被认为该Kafka server挂掉了。
            // 如果把这个值设置得过低可能被误认为挂掉，如果设置得过高，如果真的挂了，则需要很长时间才能被server得知。
            propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionOutTime);
            propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            //组与组间的消费者是没有关系的。
            //topic中已有分组消费数据，新建其他分组ID的消费者时，之前分组提交的offset对新建的分组消费不起作用。
            //propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, testGroup);
    
            //当创建一个新分组的消费者时，auto.offset.reset值为latest时，
            // 表示消费新的数据（从consumer创建开始，后生产的数据），之前产生的数据不消费。
            // https://blog.csdn.net/u012129558/article/details/80427016
    
            //earliest 当分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费。
           // latest 当分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据。
    
            propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, offset);
            //不是指每次都拉50条数据，而是一次最多拉50条数据()
            propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 5);
            return propsMap;
        }
    }

3、展示kafka消费者

@Component
    public class KafkaConsumer {
        private static final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);
    
    
        @KafkaListener(topics = "${kafka.test.topic1}",groupId = "${kafka.test.group1}",containerFactory = "kafkaListenerContainerFactory")
        public void listenPartition1(List<ConsumerRecord<?, ?>> records,Acknowledgment ack) {
            logger.info("testTopic1 recevice a message size :{}" , records.size());
    
    
            try {
                for (ConsumerRecord<?, ?> record : records) {
                    Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                    logger.info("received:{} " , record);
                    if (kafkaMessage.isPresent()) {
                        Object message = record.value();
                        String topic = record.topic();
                        Thread.sleep(300);
                        logger.info("p1 topic is:{} received message={}",topic, message);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                ack.acknowledge();
            }
    
        }
    
        @KafkaListener(topics = "${kafka.test.topic2}",groupId = "${kafka.test.group2}",containerFactory = "kafkaListenerContainerFactory")
        public void listenPartition2(List<ConsumerRecord<?, ?>> records,Acknowledgment ack) {
            logger.info("testTopic2 recevice a message size :{}" , records.size());
    
    
            try {
                for (ConsumerRecord<?, ?> record : records) {
                    Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                    logger.info("received:{} " , record);
                    if (kafkaMessage.isPresent()) {
                        Object message = record.value();
                        String topic = record.topic();
                        Thread.sleep(300);
                        logger.info("p2 topic :{},received message={}",topic, message);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                ack.acknowledge();
            }
    
        }
    
        @KafkaListener(topics = "${kafka.test.topic3}",groupId = "${kafka.test.group3}",containerFactory = "kafkaListenerContainerFactory")
        public void listenPartition3(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
            logger.info("testTopic3 recevice a message size :{}" , records.size());
    
    
            try {
                for (ConsumerRecord<?, ?> record : records) {
                    Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                    logger.info("received:{} " , record);
                    if (kafkaMessage.isPresent()) {
                        Object message = record.value();
                        String topic = record.topic();
                        logger.info("p3 topic :{},received message={}",topic, message);
                        Thread.sleep(300);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                ack.acknowledge();
            }
        }
    
    }

查看分区消费情况：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhamluZ19mZWl5dWU_size_16_color_FFFFFF_t_70]: /images/20230528/74950de5ef664a21b385c09a82ba0a3e.png