Kafka消费者负载均衡策略消费者再平衡 consumer rebalance 心跳机制

约定不等于承诺〃 2022-12-15 14:22 224阅读 0赞

# 消费者负载均衡策略 #

## 一、kafka消费 ##

Kafka提供了非常简单的消费API，使用者只需**初始化Kafka的Broker Server地址**，然后实**例化KafkaConsumer类**即可拿到Topic中的数据。一个简单的Kafka消费实例代码如下所示：

public class JConsumerSubscribe extends Thread {
        public static void main(String[] args) {
            JConsumerSubscribe jconsumer = new JConsumerSubscribe();
            jconsumer.start();
        }
    
        /** 初始化Kafka集群信息. */
        private Properties configure() {
            Properties props = new Properties();
            props.put("bootstrap.servers", "dn1:9093,dn2:9093,dn3:9093");// 指定Kafka集群地址
            props.put("group.id", "aiface");// 指定消费者组
            props.put("enable.auto.commit", "true");// 开启自动提交
            props.put("auto.commit.interval.ms", "1000");// 自动提交的时间间隔
            // 反序列化消息主键
            props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            // 反序列化消费记录
            props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            return props;
        }
    
        /** 实现一个单线程消费者. */
        @Override
        public void run() {
            // 创建一个消费者实例对象
            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(configure());
            // 订阅消费主题集合
            consumer.subscribe(Arrays.asList("aiface_topic"));
            // 实时消费标识
            boolean flag = true;
            while (flag) {
                // 获取主题消息数据
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records)
                    // 循环打印消息记录
                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
            // 出现异常关闭消费者对象
            consumer.close();
        }
    }

当我们调用**poll()方法**获取主题数据的时候，**Kafka Broker Server**做了那些事情。接下来，我们可以去看看源代码的实现细节。核心代码如下：

private ConsumerRecords<K, V> poll(final long timeoutMs, final boolean includeMetadataInTimeout) {
            acquireAndEnsureOpen();
            try {
                if (timeoutMs < 0) throw new IllegalArgumentException("Timeout must not be negative");
    
                if (this.subscriptions.hasNoSubscriptionOrUserAssignment()) {
                    throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
                }
    
                // poll for new data until the timeout expires
                long elapsedTime = 0L;
                do {
    
                    client.maybeTriggerWakeup();
    
                    final long metadataEnd;
                    if (includeMetadataInTimeout) {
                        final long metadataStart = time.milliseconds();
                        if (!updateAssignmentMetadataIfNeeded(remainingTimeAtLeastZero(timeoutMs, elapsedTime))) {
                            return ConsumerRecords.empty();
                        }
                        metadataEnd = time.milliseconds();
                        elapsedTime += metadataEnd - metadataStart;
                    } else {
                        while (!updateAssignmentMetadataIfNeeded(Long.MAX_VALUE)) {
                            log.warn("Still waiting for metadata");
                        }
                        metadataEnd = time.milliseconds();
                    }
    
                    final Map<TopicPartition, List<ConsumerRecord<K, V>>> records = pollForFetches(remainingTimeAtLeastZero(timeoutMs, elapsedTime));
    
                    if (!records.isEmpty()) {
                        // before returning the fetched records, we can send off the next round of fetches
                        // and avoid block waiting for their responses to enable pipelining while the user
                        // is handling the fetched records.
                        //
                        // NOTE: since the consumed position has already been updated, we must not allow
                        // wakeups or any other errors to be triggered prior to returning the fetched records.
                        if (fetcher.sendFetches() > 0 || client.hasPendingRequests()) {
                            client.pollNoWakeup();
                        }
    
                        return this.interceptors.onConsume(new ConsumerRecords<>(records));
                    }
                    final long fetchEnd = time.milliseconds();
                    elapsedTime += fetchEnd - metadataEnd;
    
                } while (elapsedTime < timeoutMs);
    
                return ConsumerRecords.empty();
            } finally {
                release();
            }
        }

上述代码中有个方法 records = pollForFetches(remainingTimeAtLeastZero(timeoutMs, elapsedTime))它的实现逻辑如下：

private Map<TopicPartition, List<ConsumerRecord<K, V>>> pollForFetches(final long timeoutMs) {
            final long startMs = time.milliseconds();
            long pollTimeout = Math.min(coordinator.timeToNextPoll(startMs), timeoutMs);
    
            // if data is available already, return it immediately
            final Map<TopicPartition, List<ConsumerRecord<K, V>>> records = fetcher.fetchedRecords();
            if (!records.isEmpty()) {
                return records;
            }
    
            // send any new fetches (won't resend pending fetches)
            fetcher.sendFetches();
    
            // We do not want to be stuck blocking in poll if we are missing some positions
            // since the offset lookup may be backing off after a failure
    
            // NOTE: the use of cachedSubscriptionHashAllFetchPositions means we MUST call
            // updateAssignmentMetadataIfNeeded before this method.
            if (!cachedSubscriptionHashAllFetchPositions && pollTimeout > retryBackoffMs) {
                pollTimeout = retryBackoffMs;
            }
    
            client.poll(pollTimeout, startMs, () -> {
                // since a fetch might be completed by the background thread, we need this poll condition
                // to ensure that we do not block unnecessarily in poll()
                return !fetcher.hasCompletedFetches();
            });
    
            // after the long poll, we should check whether the group needs to rebalance
            // prior to returning data so that the group can stabilize faster
            if (coordinator.rejoinNeededOrPending()) {
                return Collections.emptyMap();
            }
    
            return fetcher.fetchedRecords();
        }

可以看出每次消费者客户端拉取数据时，通过poll方法，先调用fetcher中的fetchedRecords函数，如果获取不到数据，就会发起一个新的sendFetches请求。而在消费数据的时候，每个批次从Kafka Broker Server中拉取数据是有**最大数据量限制**，默认是500条，由属性（max.poll.records）控制**，**可以在客户端中设置该属性值来调整我们消费时每次拉取数据的量。**max.poll.records返回的是一个poll请求的数据总和，与多少个分区无关**。因此，每次消费从所有分区中拉取Topic的数据的总条数不会超过max.poll.records所设置的值。

而在Fetcher的类中，在sendFetches方法中有**限制拉取数据容量的限制，由属性（max.partition.fetch.bytes），默认1MB**。可能会有这样一个场景，当满足max.partition.fetch.bytes限制条件，如果需要Fetch出5000条记录，每次默认500条，那么我们需要执行10次才能将这一次通过网络发起的请求全部Fetch完毕。

这里，可能有同学有疑问，我们不能将默认的max.poll.records属性值调到10000吗？可以调，但是还有个属性需要一起配合才可以，这个就是**每次poll的超时时间（Duration.ofMillis(100)）**，这里需要根据你的实际每条数据的容量大小来确定设置超时时间，如果你将最大值调到10000，当你每条记录的容量很大时，超时时间还是100ms，那么可能拉取的数据少于10000条。

最后还需要注意的事情，就是**会话超时的问题。session.timeout.ms默认是10s**，group.min.session.timeout.ms默认是6s，group.max.session.timeout.ms默认是30min。当你在处理消费的业务逻辑的时候，**如果在10s内没有处理完，那么消费者客户端就会与Kafka Broker Server断开，消费掉数据所产生的offset就无法提交给Kafka**，因为Kafka Broker Server此时认为该消费者程序已经断开，而即使你设置了自动提交属性，或者设置auto.offset.reset属性，你消费的时候还是会出现**重复消费**的情况，这就是因为session.timeout.ms超时的原因导致的。

## 二、消费者再平衡过程（consumer rebalance） ##

###   1. 消费者再平衡过程 ###

**消费者再平衡过程是指 消费者重新加入消费组，并且重新分配分区Partition给消费者的过程。**

###    2. 会引起消费者再平衡的情况 ###

*  新的消费者**加入**消费组
 *  某个消费者从消费组中**退出**（异常或正常）
 *  增加**订阅主题**的分区（kafka的分区数，可以动态的增加，但不能减少）
 *  某台**broker宕机**，新的协调器ZK当选
 *  某个消费者在**心跳会话时间**内没有发送心跳请求（配置参数：session.timeout.ms）, 组ZK认为该消费者已经退出。

## 三、消费者与分区Partition的对应关系 ##

**        同一个分区中的数据，只能被一个消费者组中的一个消费者所消费。例如 P0分区中的数据不能被Consumer Group A中C1与C2同时消费。**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4OTk5ODU_size_16_color_FFFFFF_t_70][]

*  如果有3个Partition, p0/p1/p2，同一个消费组有3个消费者，c0/c1/c2，则为一一对应关系；
 *  如果有3个Partition, p0/p1/p2，同一个消费组有2个消费者，c0/c1，则其中一个消费者消费2个分区的数据，另一个消费者消费一个分区的数据；
 *  如果有2个Partition, p0/p1，同一个消费组有3个消费者，c0/c1/c3，则其中有一个消费者空闲，另外2个消费者消费分别各自消费一个分区的数据；

## 四、关于消费线程与分区的对应关系 ##

理论上消费线程数应该小于等于分区数。一个消费线程对应一个分区，**当消费线程等于分区数是最大化线程的利用率**。直接使用KafkaConsumer Client实例，但是，如果我们有富裕的CPU，其实还可以使用大于分区数的消费线程，来提升消费能力，这就需要我们对KafkaConsumer Client实例进行改造，实现**消费策略预计算**，利用额外的CPU开启更多的线程，来实现**消费任务分片**，有效解决消息挤压 、消息队列满等问题。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA4OTk5ODU_size_16_color_FFFFFF_t_70]: /images/20221123/e1438519131642ebb6e8fe5693131fa4.png