1.概述

最近有同學(xué)咨詢(xún)Kafka的消費(fèi)和心跳機(jī)制，今天筆者將通過(guò)這篇博客來(lái)逐一介紹這些內(nèi)容。

2.內(nèi)容

2.1 Kafka消費(fèi)

首先，我們來(lái)看看消費(fèi)。Kafka提供了非常簡(jiǎn)單的消費(fèi)API，使用者只需初始化Kafka的Broker Server地址，然后實(shí)例化KafkaConsumer類(lèi)即可拿到Topic中的數(shù)據(jù)。一個(gè)簡(jiǎn)單的Kafka消費(fèi)實(shí)例代碼如下所示：

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:9092,dn2:9092,dn3:9092");// 指定Kafka集群地址 
        props.put("group.id", "ke");// 指定消費(fèi)者組 
        props.put("enable.auto.commit", "true");// 開(kāi)啟自動(dòng)提交 
        props.put("auto.commit.interval.ms", "1000");// 自動(dòng)提交的時(shí)間間隔 
        // 反序列化消息主鍵        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); 
        // 反序列化消費(fèi)記錄        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); 
        return props; 
    }    /** 實(shí)現(xiàn)一個(gè)單線(xiàn)程消費(fèi)者. */    @Override    public void run() {        // 創(chuàng)建一個(gè)消費(fèi)者實(shí)例對(duì)象        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(configure());        // 訂閱消費(fèi)主題集合        consumer.subscribe(Arrays.asList("test_kafka_topic")); 
        // 實(shí)時(shí)消費(fèi)標(biāo)識(shí)        boolean flag = true; 
        while (flag) { 
            // 獲取主題消息數(shù)據(jù)            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); 
            for (ConsumerRecord<String, String> record : records) 
                // 循環(huán)打印消息記錄                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); 
        }        // 出現(xiàn)異常關(guān)閉消費(fèi)者對(duì)象        consumer.close(); 
    }} 

上述代碼我們就可以非常便捷的拿到Topic中的數(shù)據(jù)。但是，當(dāng)我們調(diào)用poll方法拉取數(shù)據(jù)的時(shí)候，Kafka Broker Server做了那些事情。接下來(lái)，我們可以去看看源代碼的實(shí)現(xiàn)細(xì)節(jié)。核心代碼如下：

org.apache.kafka.clients.consumer.KafkaConsumer

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();        }    } 

上述代碼中有個(gè)方法pollForFetches，它的實(shí)現(xiàn)邏輯如下：

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(); 
    } 

上述代碼中加粗的位置，我們可以看出每次消費(fèi)者客戶(hù)端拉取數(shù)據(jù)時(shí)，通過(guò)poll方法，先調(diào)用fetcher中的fetchedRecords函數(shù)，如果獲取不到數(shù)據(jù)，就會(huì)發(fā)起一個(gè)新的sendFetches請(qǐng)求。而在消費(fèi)數(shù)據(jù)的時(shí)候，每個(gè)批次從Kafka Broker Server中拉取數(shù)據(jù)是有最大數(shù)據(jù)量限制，默認(rèn)是500條，由屬性(max.poll.records)控制，可以在客戶(hù)端中設(shè)置該屬性值來(lái)調(diào)整我們消費(fèi)時(shí)每次拉取數(shù)據(jù)的量。

提示：這里需要注意的是，max.poll.records返回的是一個(gè)poll請(qǐng)求的數(shù)據(jù)總和，與多少個(gè)分區(qū)無(wú)關(guān)。因此，每次消費(fèi)從所有分區(qū)中拉取Topic的數(shù)據(jù)的總條數(shù)不會(huì)超過(guò)max.poll.records所設(shè)置的值。

而在Fetcher的類(lèi)中，在sendFetches方法中有限制拉取數(shù)據(jù)容量的限制，由屬性(max.partition.fetch.bytes)，默認(rèn)1MB?？赡軙?huì)有這樣一個(gè)場(chǎng)景，當(dāng)滿(mǎn)足max.partition.fetch.bytes限制條件，如果需要Fetch出10000條記錄，每次默認(rèn)500條，那么我們需要執(zhí)行20次才能將這一次通過(guò)網(wǎng)絡(luò)發(fā)起的請(qǐng)求全部Fetch完畢。

這里，可能有同學(xué)有疑問(wèn)，我們不能將默認(rèn)的max.poll.records屬性值調(diào)到10000嗎?可以調(diào)，但是還有個(gè)屬性需要一起配合才可以，這個(gè)就是每次poll的超時(shí)時(shí)間(Duration.ofMillis(100))，這里需要根據(jù)你的實(shí)際每條數(shù)據(jù)的容量大小來(lái)確定設(shè)置超時(shí)時(shí)間，如果你將最大值調(diào)到10000，當(dāng)你每條記錄的容量很大時(shí)，超時(shí)時(shí)間還是100ms，那么可能拉取的數(shù)據(jù)少于10000條。

而這里，還有另外一個(gè)需要注意的事情，就是會(huì)話(huà)超時(shí)的問(wèn)題。session.timeout.ms默認(rèn)是10s，group.min.session.timeout.ms默認(rèn)是6s，group.max.session.timeout.ms默認(rèn)是30min。當(dāng)你在處理消費(fèi)的業(yè)務(wù)邏輯的時(shí)候，如果在10s內(nèi)沒(méi)有處理完，那么消費(fèi)者客戶(hù)端就會(huì)與Kafka Broker Server斷開(kāi)，消費(fèi)掉的數(shù)據(jù)，產(chǎn)生的offset就沒(méi)法提交給Kafka，因?yàn)镵afka Broker Server此時(shí)認(rèn)為該消費(fèi)者程序已經(jīng)斷開(kāi)，而即使你設(shè)置了自動(dòng)提交屬性，或者設(shè)置auto.offset.reset屬性，你消費(fèi)的時(shí)候還是會(huì)出現(xiàn)重復(fù)消費(fèi)的情況，這就是因?yàn)閟ession.timeout.ms超時(shí)的原因?qū)е碌摹?/p>

2.2 心跳機(jī)制

上面在末尾的時(shí)候，說(shuō)到會(huì)話(huà)超時(shí)的情況導(dǎo)致消息重復(fù)消費(fèi)，為什么會(huì)有超時(shí)?有同學(xué)會(huì)有這樣的疑問(wèn)，我的消費(fèi)者線(xiàn)程明明是啟動(dòng)的，也沒(méi)有退出，為啥消費(fèi)不到Kafka的消息呢?消費(fèi)者組也查不到我的ConsumerGroupID呢?這就有可能是超時(shí)導(dǎo)致的，而Kafka是通過(guò)心跳機(jī)制來(lái)控制超時(shí)，心跳機(jī)制對(duì)于消費(fèi)者客戶(hù)端來(lái)說(shuō)是無(wú)感的，它是一個(gè)異步線(xiàn)程，當(dāng)我們啟動(dòng)一個(gè)消費(fèi)者實(shí)例時(shí)，心跳線(xiàn)程就開(kāi)始工作了。

在org.apache.kafka.clients.consumer.internals.AbstractCoordinator中會(huì)啟動(dòng)一個(gè)HeartbeatThread線(xiàn)程來(lái)定時(shí)發(fā)送心跳和檢測(cè)消費(fèi)者的狀態(tài)。每個(gè)消費(fèi)者都有個(gè)org.apache.kafka.clients.consumer.internals.ConsumerCoordinator，而每個(gè)ConsumerCoordinator都會(huì)啟動(dòng)一個(gè)HeartbeatThread線(xiàn)程來(lái)維護(hù)心跳，心跳信息存放在org.apache.kafka.clients.consumer.internals.Heartbeat中，聲明的Schema如下所示：

private final int sessionTimeoutMs; 
    private final int heartbeatIntervalMs; 
    private final int maxPollIntervalMs; 
    private final long retryBackoffMs; 
    private volatile long lastHeartbeatSend;  
    private long lastHeartbeatReceive; 
    private long lastSessionReset; 
    private long lastPoll; 
    private boolean heartbeatFailed; 

心跳線(xiàn)程中的run方法實(shí)現(xiàn)代碼如下：

public void run() { 
            try { 
                log.debug("Heartbeat thread started"); 
                while (true) { 
                    synchronized (AbstractCoordinator.this) { 
                        if (closed) 
                            return; 
                        if (!enabled) { 
                            AbstractCoordinator.this.wait(); 
                            continue; 
                        }                        if (state != MemberState.STABLE) { 
                            // the group is not stable (perhaps because we left the group or because the coordinator 
                            // kicked us out), so disable heartbeats and wait for the main thread to rejoin. 
                            disable(); 
                            continue; 
                        } 
                        client.pollNoWakeup(); 
                        long now = time.milliseconds(); 
                        if (coordinatorUnknown()) { 
                            if (findCoordinatorFuture != null || lookupCoordinator().failed()) 
                                // the immediate future check ensures that we backoff properly in the case that no 
                                // brokers are available to connect to. 
                                AbstractCoordinator.this.wait(retryBackoffMs); 
                        } else if (heartbeat.sessionTimeoutExpired(now)) { 
                            // the session timeout has expired without seeing a successful heartbeat, so we should 
                            // probably make sure the coordinator is still healthy. 
                            markCoordinatorUnknown(); 
                        } else if (heartbeat.pollTimeoutExpired(now)) { 
                            // the poll timeout has expired, which means that the foreground thread has stalled 
                            // in between calls to poll(), so we explicitly leave the group. 
                            maybeLeaveGroup(); 
                        } else if (!heartbeat.shouldHeartbeat(now)) { 
                            // poll again after waiting for the retry backoff in case the heartbeat failed or the 
                            // coordinator disconnected 
                            AbstractCoordinator.this.wait(retryBackoffMs); 
                        } else { 
                            heartbeat.sentHeartbeat(now); 
                            sendHeartbeatRequest().addListener(new RequestFutureListener<Void>() { 
                                @Override 
                                public void onSuccess(Void value) { 
                                    synchronized (AbstractCoordinator.this) { 
                                        heartbeat.receiveHeartbeat(time.milliseconds()); 
                                    } 
                                } 
                                @Override 
                                public void onFailure(RuntimeException e) { 
                                    synchronized (AbstractCoordinator.this) { 
                                        if (e instanceof RebalanceInProgressException) { 
                                            // it is valid to continue heartbeating while the group is rebalancing. This 
                                            // ensures that the coordinator keeps the member in the group for as long 
                                            // as the duration of the rebalance timeout. If we stop sending heartbeats, 
                                            // however, then the session timeout may expire before we can rejoin. 
                                            heartbeat.receiveHeartbeat(time.milliseconds()); 
                                        } else { 
                                            heartbeat.failHeartbeat(); 
                                            // wake up the thread if it's sleeping to reschedule the heartbeat 
                                            AbstractCoordinator.this.notify(); 
                                        } 
                                    } 
                                } 
                            }); 
                        } 
                    } 
                } 
            } catch (AuthenticationException e) { 
                log.error("An authentication error occurred in the heartbeat thread", e); 
                this.failed.set(e); 
            } catch (GroupAuthorizationException e) { 
                log.error("A group authorization error occurred in the heartbeat thread", e); 
                this.failed.set(e); 
            } catch (InterruptedException | InterruptException e) { 
                Thread.interrupted(); 
                log.error("Unexpected interrupt received in heartbeat thread", e); 
                this.failed.set(new RuntimeException(e)); 
            } catch (Throwable e) { 
                log.error("Heartbeat thread failed due to unexpected error", e); 
                if (e instanceof RuntimeException) 
                    this.failed.set((RuntimeException) e); 
                else 
                    this.failed.set(new RuntimeException(e)); 
            } finally { 
                log.debug("Heartbeat thread has closed"); 
            } 
        } 
View Code 

在心跳線(xiàn)程中這里面包含兩個(gè)最重要的超時(shí)函數(shù)，它們是sessionTimeoutExpired和pollTimeoutExpired。

public boolean sessionTimeoutExpired(long now) { 
        return now - Math.max(lastSessionReset, lastHeartbeatReceive) > sessionTimeoutMs; 
}public boolean pollTimeoutExpired(long now) { 
        return now - lastPoll > maxPollIntervalMs; 
} 

2.2.1 sessionTimeoutExpired

如果是sessionTimeout超時(shí)，則會(huì)被標(biāo)記為當(dāng)前協(xié)調(diào)器處理斷開(kāi)，此時(shí)，會(huì)將消費(fèi)者移除，重新分配分區(qū)和消費(fèi)者的對(duì)應(yīng)關(guān)系。在Kafka Broker Server中，Consumer Group定義了5中(如果算上Unknown，應(yīng)該是6種狀態(tài))狀態(tài)，org.apache.kafka.common.ConsumerGroupState，如下圖所示：

2.2.2 pollTimeoutExpired

如果觸發(fā)了poll超時(shí)，此時(shí)消費(fèi)者客戶(hù)端會(huì)退出ConsumerGroup，當(dāng)再次poll的時(shí)候，會(huì)重新加入到ConsumerGroup，觸發(fā)RebalanceGroup。而KafkaConsumer Client是不會(huì)幫我們重復(fù)poll的，需要我們自己在實(shí)現(xiàn)的消費(fèi)邏輯中不停的調(diào)用poll方法。

3.分區(qū)與消費(fèi)線(xiàn)程

關(guān)于消費(fèi)分區(qū)與消費(fèi)線(xiàn)程的對(duì)應(yīng)關(guān)系，理論上消費(fèi)線(xiàn)程數(shù)應(yīng)該小于等于分區(qū)數(shù)。之前是有這樣一種觀(guān)點(diǎn)，一個(gè)消費(fèi)線(xiàn)程對(duì)應(yīng)一個(gè)分區(qū)，當(dāng)消費(fèi)線(xiàn)程等于分區(qū)數(shù)是最大化線(xiàn)程的利用率。直接使用KafkaConsumer Client實(shí)例，這樣使用確實(shí)沒(méi)有什么問(wèn)題。但是，如果我們有富裕的CPU，其實(shí)還可以使用大于分區(qū)數(shù)的線(xiàn)程，來(lái)提升消費(fèi)能力，這就需要我們對(duì)KafkaConsumer Client實(shí)例進(jìn)行改造，實(shí)現(xiàn)消費(fèi)策略預(yù)計(jì)算，利用額外的CPU開(kāi)啟更多的線(xiàn)程，來(lái)實(shí)現(xiàn)消費(fèi)任務(wù)分片。具體實(shí)現(xiàn)，留到下一篇博客，給大家分享《基于Kafka的分布式查詢(xún)SQL引擎》。

4.結(jié)束語(yǔ)

這篇博客就和大家分享到這里，如果大家在研究學(xué)習(xí)的過(guò)程當(dāng)中有什么問(wèn)題，可以加群進(jìn)行討論或發(fā)送郵件給我，我會(huì)盡我所能為您解答，與君共勉!