解決kafka消息堆積及分區不均勻的問題

kafka消息堆積及分區不均勻的解決

我在環境中發現代碼裡面的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;
    }
}

4、展示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();
        }
    }
}

查看分區消費情況:

在這裡插入圖片描述

kafka出現若幹分區不消費的現象

近日,有用戶反饋kafka有topic出現某個消費組消費的時候,有幾個分區一直不消費消息,消息一直積壓(圖1)。除瞭一直積壓外,還有一個現象就是消費組一直在重均衡,大約每5分鐘就會重均衡一次。具體表現為消費分區的owner一直在改變(圖2)。

(圖1)

(圖2)

定位過程

業務側沒有報錯,同時kafka服務端日志也一切正常,同事先將消費組的機器滾動重啟,仍然還是那幾個分區沒有消費,之後將這幾個不消費的分區遷移至別的broker上,依然沒有消費。

還有一個奇怪的地方,就是每次重均衡後,不消費的那幾個分區的消費owner所在機器的網絡都有流量變化。按理說不消費應該就是拉取不到分區不會有流量的。於是讓運維去拉瞭下不消費的consumer的jstack日志。一看果然發現瞭問題所在。

從堆棧看,consumer已經拉取到消息,然後就一直卡在處理消息的業務邏輯上。這說明kafka是沒有問題的,用戶的業務邏輯有問題。

consumer在拉取完一批消息後,就一直在處理這批消息,但是這批消息中有若幹條消息無法處理,而業務又沒有超時操作或者異常處理導致進程一直處於消費中,無法去poll下一批數據。

又由於業務采用的是autocommit的offset提交方式,而根據源碼可知,consumer隻有在下一次poll中才會自動提交上次poll的offset,所以業務一直在拉取同一批消息而無法更新offset。反映的現象就是該consumer對應的分區的offset一直沒有變,所以有積壓的現象。

至於為什麼會一直在重均衡消費組的原因也很明瞭瞭,就是因為有消費者一直卡在處理消息的業務邏輯上,超過瞭max.poll.interval.ms(默認5min),消費組就會將該消費者踢出消費組,從而發生重均衡。

驗證

讓業務方去查證業務日志,驗證瞭積壓的這幾個分區,總是在循環的拉取同一批消息。

解決方法

臨時解決方法就是跳過有問題的消息,將offset重置到有問題的消息之後。本質上還是要業務側修改業務邏輯,增加超時或者異常處理機制,最好不要采用自動提交offset的方式,可以手動管理。

以上為個人經驗,希望能給大傢一個參考,也希望大傢多多支持WalkonNet。

推薦閱讀: