您的位置：首页 > 理论基础 > 数据结构算法

Kafka源码分析Producer读取Metadata的数据结构及Metadata两种更新机制介绍

2016-09-29 00:00 387 查看

问题导读：
1.什么是多线程异步发送模型？
2.Metadata的线程安全性如何实现？
3.Metadata的数据结构是什么？
4.producer如何读取Metadata？
5.Sender的如何创建?
6.Senderpoll()如何更新Metadata？
7.Metadata有哪2种更新机制?
8.什么是Metadata失效检测?
9.Metadata有哪些其他的更新策略？

解决方案：

多线程异步发送模型

下图是经过源码分析之后，整理出来的Producer端的架构图：

在上一篇我们讲过，Producer有同步发送和异步发送2种策略。在以前的Kafkaclientapi实现中，同步和异步是分开实现的。而在0.9中，同步发送其实是通过异步发送间接实现，其接口如下：

1
2
3
4
5
6
7

publicclassKafkaProducer<K,V>implementsProducer<K,V>{
...
publicFuture<RecordMetadata>send(ProducerRecord<K,V>record,Callbackcallback)//异步发送接口
{
...
}
}

要实现同步发送，只要在拿到返回的Future对象之后，直接调用get()就可以了。

基本思路

从上图我们可以看出，异步发送的基本思路就是：send的时候，KafkaProducer把消息放到本地的消息队列RecordAccumulator，然后一个后台线程Sender不断循环，把消息发给Kafka集群。

要实现这个，还得有一个前提条件：就是KafkaProducer/Sender都需要获取集群的配置信息Metadata。所谓Metadata，也就是在上一篇所讲的，Topic/Partion与broker的映射关系：每一个Topic的每一个Partion，得知道其对应的broker列表是什么，其中leader是谁，follower是谁。

2个数据流

所以在上图中，有2个数据流：
Metadata流(A1,A2,A3)：Sender从集群获取信息，然后更新Metadata；KafkaProducer先读取Metadata，然后把消息放入队列。

消息流(B1,B2,B3)：这个很好理解，不再详述。

本篇着重讲述Metadata流，消息流，将在后续详细讲述。

Metadata的线程安全性

从上图可以看出，Metadata是多个producer线程读，一个sender线程更新，因此它必须是线程安全的。

Kafka的官方文档上也有说明，KafkaProducer是线程安全的，可以在多线程中调用：

Theproduceristhreadsafeandsharingasingleproducerinstanceacrossthreadswillgenerallybefasterthanhavingmultipleinstances.

从下面代码也可以看出，它的所有public方法都是synchronized:

01
02
03
04
05
06
07
08
09
10
11
12
13

publicfinalclassMetadata{
。。。
publicsynchronizedClusterfetch(){
returnthis.cluster;
}
publicsynchronizedlongtimeToNextUpdate(longnowMs){
。。。
}
publicsynchronizedintrequestUpdate(){
。。。
}
。。。
｝

Metadata的数据结构

下面代码列举了Metadata的主要数据结构：一个Cluster对象+1堆状态变量。前者记录了集群的配置信息，后者用于控制Metadata的更新策略。

01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

publicfinalclassMetadata{
...
privatefinallongrefreshBackoffMs;//更新失败的情况下，下1次更新的补偿时间（这个变量在代码中意义不是太大）
privatefinallongmetadataExpireMs;//关键值：每隔多久，更新一次。缺省是600*1000，也就是10分种
privateintversion;//每更新成功1次，version递增1。这个变量主要用于在while循环，wait的时候，作为循环判断条件
privatelonglastRefreshMs;//上一次更新时间（也包含更新失败的情况）
privatelonglastSuccessfulRefreshMs;//上一次成功更新的时间（如果每次都成功的话，则2者相等。否则，lastSuccessulRefreshMs<lastRefreshMs)
privateClustercluster;//集群配置信息
privatebooleanneedUpdate;//是否强制刷新
、
...
}

publicfinalclassCluster{
...
privatefinalList<Node>nodes;//Node也就是Broker
privatefinalMap<TopicPartition,PartitionInfo>partitionsByTopicPartition;//Topic/Partion和brokerlist的映射关系
privatefinalMap<String,List<PartitionInfo>>partitionsByTopic;
privatefinalMap<String,List<PartitionInfo>>availablePartitionsByTopic;
privatefinalMap<Integer,List<PartitionInfo>>partitionsByNode;
privatefinalMap<Integer,Node>nodesById;
}

publicclassPartitionInfo{
privatefinalStringtopic;
privatefinalintpartition;
privatefinalNodeleader;
privatefinalNode[]replicas;
privatefinalNode[]inSyncReplicas;
}

producer读取Metadata

下面是send函数的源码，可以看到，在send之前，会先读取metadata。如果metadata读不到，会一直阻塞在那，直到超时，抛出TimeoutException

01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

//KafkaProducer
publicFuture<RecordMetadata>send(ProducerRecord<K,V>record,Callbackcallback){
try{
longwaitedOnMetadataMs=waitOnMetadata(record.topic(),this.maxBlockTimeMs);//拿不到topic的配置信息，会一直阻塞在这，直到抛异常

...//拿到了，执行下面的send逻辑
}catch()
{}
}

//KafkaProducer
privatelongwaitOnMetadata(Stringtopic,longmaxWaitMs)throwsInterruptedException{
if(!this.metadata.containsTopic(topic))
this.metadata.add(topic);

if(metadata.fetch().partitionsForTopic(topic)!=null)
return0;//取到topic的配置信息，直接返回

longbegin=time.milliseconds();
longremainingWaitMs=maxWaitMs;
while(metadata.fetch().partitionsForTopic(topic)==null){//取不到topic的配置信息，一直死循环wait，直到超时，抛TimeoutException
log.trace("Requestingmetadataupdatefortopic{}.",topic);
intversion=metadata.requestUpdate();//把needUpdate置为true
sender.wakeup();//唤起sender

metadata.awaitUpdate(version,remainingWaitMs);//metadata的关键函数
longelapsed=time.milliseconds()-begin;
if(elapsed>=maxWaitMs)
thrownewTimeoutException("Failedtoupdatemetadataafter"+maxWaitMs+"ms.");
if(metadata.fetch().unauthorizedTopics().contains(topic))
thrownewTopicAuthorizationException(topic);
remainingWaitMs=maxWaitMs-elapsed;
}
returntime.milliseconds()-begin;
}

//Metadata
publicsynchronizedvoidawaitUpdate(finalintlastVersion,finallongmaxWaitMs)throwsInterruptedException{
if(maxWaitMs<0){
thrownewIllegalArgumentException("Maxtimetowaitformetadataupdatesshouldnotbe<0milliseconds");
}
longbegin=System.currentTimeMillis();
longremainingWaitMs=maxWaitMs;
while(this.version<=lastVersion){//当Sender成功更新meatadata之后，version加1。否则会循环，一直wait
if(remainingWaitMs!=0
wait(remainingWaitMs);//线程的wait机制，wait和synchronized的配合使用
longelapsed=System.currentTimeMillis()-begin;
if(elapsed>=maxWaitMs)//wait时间超出了最长等待时间
thrownewTimeoutException("Failedtoupdatemetadataafter"+maxWaitMs+"ms.");
remainingWaitMs=maxWaitMs-elapsed;
}
}

总结：从上面代码可以看出，producerwaitmetadata的时候，有2个条件：

（1）while(metadata.fetch().partitionsForTopic(topic)==null)

（2）while(this.version<=lastVersion)

有wait就会有notify，notify在Sender更新Metadata的时候发出。

Sender的创建

下面是KafkaProducer的构造函数，从代码可以看出，Sender就是KafkaProducer中创建的一个Thread.

01
02
03
04
05
06
07
08
09
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

privateKafkaProducer(ProducerConfigconfig,Serializer<K>keySerializer,Serializer<V>valueSerializer){
try{
...
this.metadata=newMetadata(retryBackoffMs,config.getLong(ProducerConfig.METADATA_MAX_AGE_CONFIG));//构造metadata

this.metadata.update(Cluster.bootstrap(addresses),time.milliseconds());//往metadata中，填入初始的，配置的node列表

ChannelBuilderchannelBuilder=ClientUtils.createChannelBuilder(config.values());

NetworkClientclient=newNetworkClient(
newSelector(config.getLong(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG),this.metrics,time,"producer",metricTags,channelBuilder),
this.metadata,
clientId,
config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION),
config.getLong(ProducerConfig.RECONNECT_BACKOFF_MS_CONFIG),
config.getInt(ProducerConfig.SEND_BUFFER_CONFIG),
config.getInt(ProducerConfig.RECEIVE_BUFFER_CONFIG),

this.sender=newSender(client,//构造一个sender。sender本身实现的是Runnable接口
this.metadata,
this.accumulator,
config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG),
(short)parseAcks(config.getString(ProducerConfig.ACKS_CONFIG)),
config.getInt(ProducerConfig.RETRIES_CONFIG),
this.metrics,
newSystemTime(),
clientId,
this.requestTimeoutMs);

StringioThreadName="kafka-producer-network-thread"+(clientId.length()>0?"|"+clientId:"");
this.ioThread=newKafkaThread(ioThreadName,this.sender,true);
this.ioThread.start();//一个线程，开启sender

Senderpoll()更新Metadata

001
002
003
004
005
006
007
008
009
010
011
012
013
014
015
016
017
018
019
020
021
022
023
024
025
026
027
028
029
030
031
032
033
034
035
036
037
038
039
040
041
042
043
044
045
046
047
048
049
050
051
052
053
054
055
056
057
058
059
060
061
062
063
064
065
066
067
068
069
070
071
072
073
074
075
076
077
078
079
080
081
082
083
084
085
086
087
088
089
090
091
092
093
094
095
096
097
098
099
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171

publicvoidrun(){
//mainloop,runsuntilcloseiscalled
while(running){
try{
run(time.milliseconds());
}catch(Exceptione){
log.error("UncaughterrorinkafkaproducerI/Othread:",e);
}
}
。。。
}

publicvoidrun(longnow){
Clustercluster=metadata.fetch();
。。。
RecordAccumulator.ReadyCheckResultresult=this.accumulator.ready(cluster,now);//遍历消息队列中所有的消息，找出对应的，已经ready的Node

if(result.unknownLeadersExist)//如果一个ready的node都没有，请求更新metadata
this.metadata.requestUpdate();

。。。

//client的2个关键函数，一个发送ClientRequest，一个接收ClientResponse。底层调用的是NIO的poll。关于nio,后面会详细介绍
for(ClientRequestrequest:requests)
client.send(request,now);

this.client.poll(pollTimeout,now);
}

//NetworkClient
publicList<ClientResponse>poll(longtimeout,longnow){
longmetadataTimeout=metadataUpdater.maybeUpdate(now);//关键点：每次poll的时候判断是否要更新metadata

try{
this.selector.poll(Utils.min(timeout,metadataTimeout,requestTimeoutMs));
}catch(IOExceptione){
log.error("UnexpectederrorduringI/O",e);
}

//processcompletedactions
longupdatedNow=this.time.milliseconds();
List<ClientResponse>responses=newArrayList<>();
handleCompletedSends(responses,updatedNow);
handleCompletedReceives(responses,updatedNow);//在返回的handler中，会处理metadata的更新
handleDisconnections(responses,updatedNow);
handleConnections();
handleTimedOutRequests(responses,updatedNow);

//invokecallbacks
for(ClientResponseresponse:responses){
if(response.request().hasCallback()){
try{
response.request().callback().onComplete(response);
}catch(Exceptione){
log.error("Uncaughterrorinrequestcompletion:",e);
}
}
}

returnresponses;
}

//DefaultMetadataUpdater
@Override
publiclongmaybeUpdate(longnow){
//shouldweupdateourmetadata?
longtimeToNextMetadataUpdate=metadata.timeToNextUpdate(now);
longtimeToNextReconnectAttempt=Math.max(this.lastNoNodeAvailableMs+metadata.refreshBackoff()-now,0);
longwaitForMetadataFetch=this.metadataFetchInProgress?Integer.MAX_VALUE:0;
//ifthereisnonodeavailabletoconnect,backoffrefreshingmetadata
longmetadataTimeout=Math.max(Math.max(timeToNextMetadataUpdate,timeToNextReconnectAttempt),
waitForMetadataFetch);

if(metadataTimeout==0){
//highlydependentonthebehaviorofleastLoadedNode.
Nodenode=leastLoadedNode(now);//找到负载最小的Node
maybeUpdate(now,node);//把更新Metadata的请求，发给这个Node
}

returnmetadataTimeout;
}

privatevoidmaybeUpdate(longnow,Nodenode){
if(node==null){
log.debug("Giveupsendingmetadatarequestsincenonodeisavailable");
//markthetimestampfornonodeavailabletoconnect
this.lastNoNodeAvailableMs=now;
return;
}
StringnodeConnectionId=node.idString();

if(canSendRequest(nodeConnectionId)){
Set<String>topics=metadata.needMetadataForAllTopics()?newHashSet<String>():metadata.topics();
this.metadataFetchInProgress=true;
ClientRequestmetadataRequest=request(now,nodeConnectionId,topics);//关键点：发送更新Metadata的Request
log.debug("Sendingmetadatarequest{}tonode{}",metadataRequest,node.id());
doSend(metadataRequest,now);//这里只是异步发送，返回的response在上面的handleCompletedReceives里面处理
}elseif(connectionStates.canConnect(nodeConnectionId,now)){
log.debug("Initializeconnectiontonode{}forsendingmetadatarequest",node.id());
initiateConnect(node,now);

}else{//connected,butcan'tsendmoreORconnecting
this.lastNoNodeAvailableMs=now;
}
}

privatevoidhandleCompletedReceives(List<ClientResponse>responses,longnow){
for(NetworkReceivereceive:this.selector.completedReceives()){
Stringsource=receive.source();
ClientRequestreq=inFlightRequests.completeNext(source);
ResponseHeaderheader=ResponseHeader.parse(receive.payload());
//Alwaysexpecttheresponseversionidtobethesameastherequestversionid
shortapiKey=req.request().header().apiKey();
shortapiVer=req.request().header().apiVersion();
Structbody=(Struct)ProtoUtils.responseSchema(apiKey,apiVer).read(receive.payload());
correlate(req.request().header(),header);
if(!metadataUpdater.maybeHandleCompletedReceive(req,now,body))
responses.add(newClientResponse(req,now,false,body));
}
}

@Override
publicbooleanmaybeHandleCompletedReceive(ClientRequestreq,longnow,Structbody){
shortapiKey=req.request().header().apiKey();
if(apiKey==ApiKeys.METADATA.id&&req.isInitiatedByNetworkClient()){
handleResponse(req.request().header(),body,now);
returntrue;
}
returnfalse;
}

//关键函数
privatevoidhandleResponse(RequestHeaderheader,Structbody,longnow){
this.metadataFetchInProgress=false;
MetadataResponseresponse=newMetadataResponse(body);
Clustercluster=response.cluster();//从response中，拿到一个新的cluster对象
if(response.errors().size()>0){
log.warn("Errorwhilefetchingmetadatawithcorrelationid{}:{}",header.correlationId(),response.errors());
}

if(cluster.nodes().size()>0){
this.metadata.update(cluster,now);//更新metadata，用新的cluster覆盖旧的cluster
}else{

从上面可以看出，Metadata的更新，是在while循环，每次调用client.poll()的时候更新的。

更新机制又有以下2种：

Metadata的2种更新机制

(1)周期性的更新:每隔一段时间更新一次，这个通过Metadata的lastRefreshMs,lastSuccessfulRefreshMs这2个字段来实现

对应的ProducerConfig配置项为：

metadata.max.age.ms//缺省300000，即10分钟1次

(2)失效检测，强制更新：检查到metadata失效以后，调用metadata.requestUpdate()强制更新。requestUpdate()函数里面其实什么都没做，就是把needUpdate置成了false

每次poll的时候，都检查这2种更新机制，达到了，就触发更新。

那如何判定Metadata失效了呢？这个在代码中很分散，有很多地方，会判定Metadata失效。

Metadata失效检测

条件1：initConnect的时候

01
02
03
04
05
06
07
08
09
10
11
12
13
14
15

privatevoidinitiateConnect(Nodenode,longnow){
StringnodeConnectionId=node.idString();
try{
log.debug("Initiatingconnectiontonode{}at{}:{}.",node.id(),node.host(),node.port());
this.connectionStates.connecting(nodeConnectionId,now);
selector.connect(nodeConnectionId,
newInetSocketAddress(node.host(),node.port()),
this.socketSendBuffer,
this.socketReceiveBuffer);
}catch(IOExceptione){
connectionStates.disconnected(nodeConnectionId,now);
metadataUpdater.requestUpdate();//判定metadata失效
log.debug("Errorconnectingtonode{}at{}:{}:",node.id(),node.host(),node.port(),e);
}
}

条件2：poll里面IO的时候，连接断掉了

1
2
3
4
5
6
7
8

privatevoidhandleDisconnections(List<ClientResponse>responses,longnow){
for(Stringnode:this.selector.disconnected()){
log.debug("Node{}disconnected.",node);
processDisconnection(responses,node,now);
}
if(this.selector.disconnected().size()>0)
metadataUpdater.requestUpdate();//判定metadata失效
}

条件3：有请求超时

01
02
03
04
05
06
07
08
09
10
11

privatevoidhandleTimedOutRequests(List<ClientResponse>responses,longnow){
List<String>nodeIds=this.inFlightRequests.getNodesWithTimedOutRequests(now,this.requestTimeoutMs);
for(StringnodeId:nodeIds){
this.selector.close(nodeId);
log.debug("Disconnectingfromnode{}duetorequesttimeout.",nodeId);
processDisconnection(responses,nodeId,now);
}

if(nodeIds.size()>0)
metadataUpdater.requestUpdate();//判定metadata失效
}

条件4：发消息的时候，有partition的leader没找到

1
2
3
4
5
6

publicvoidrun(longnow){
Clustercluster=metadata.fetch();
RecordAccumulator.ReadyCheckResultresult=this.accumulator.ready(cluster,now);

if(result.unknownLeadersExist)
this.metadata.requestUpdate();

条件5：返回的response和请求对不上的时候

1
2
3
4
5
6
7
8

privatevoidhandleProduceResponse(ClientResponseresponse,Map<TopicPartition,RecordBatch>batches,longnow){
intcorrelationId=response.request().request().header().correlationId();
if(response.wasDisconnected()){
log.trace("Cancelledrequest{}duetonode{}beingdisconnected",response,response.request()
.request()
.destination());
for(RecordBatchbatch:batches.values())
completeBatch(batch,Errors.NETWORK_EXCEPTION,-1L,correlationId,now);

总之1句话：发生各式各样的异常，数据不同步，都认为metadata可能出问题了，要求更新。

Metadata其他的更新策略

除了上面所述，Metadata的更新，还有以下几个特点：

1.更新请求MetadataRequest是nio异步发送的，在poll的返回中，处理MetadataResponse的时候，才真正更新Metadata。

这里有个关键点：Metadata的cluster对象，每次是整个覆盖的，而不是局部更新。所以cluster内部不用加锁。

2.更新的时候，是从metadata保存的所有Node，或者说Broker中，选负载最小的那个，也就是当前接收请求最少的那个。向其发送MetadataRequest请求，获取新的Cluster对象。

文章转自About云(http://www.aboutyun.com/thread-19917-1-1.html)，原文位于csdn，作者：travi

内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理

标签： kafka metadata

相关文章推荐

新的分享

章节导航