Kafka producer原理 (Scala版同步producer)
Posted the_tops ----
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Kafka producer原理 (Scala版同步producer)相关的知识,希望对你有一定的参考价值。
转载自:http://www.cnblogs.com/huxi2b/p/4583249.html 供参考
本文分析的Kafka代码为kafka-0.8.2.1。另外,由于Kafka目前提供了两套Producer代码,一套是Scala版的旧版本;一套是Java版的新版本。虽然Kafka社区极力推荐大家使用Java版本的producer,但目前很多已有的程序还是调用了Scala版的API。今天我们就分析一下旧版producer的代码。
1
2
3
4
5
|
val requestRequiredAcksOpt = parser.accepts( "request-required-acks" , "The required acks of the producer requests" ) .withRequiredArg .describedAs( "request required acks" ) .ofType(classOf[java.lang.Integer]) .defaultsTo( 0 ) // 此处默认设置为0 |
1
2
3
4
5
|
do { message = reader.readMessage() // 从LineMessageReader类中读取消息。该类接收键盘输入的一行文本作为消息 if (message ! = null ) producer.send(message.topic, message.key, message.message) // key默认是空,如果想要指定,需传入参数parse.key=true,默认key和消息文本之间的分隔符是\'\\t\' } while (message ! = null ) // 循环接收消息,除非Ctrl+C或其他其他引发IOException操作跳出循环 |
下面代码是Producer.scala中的发送方法:
1
2
3
4
5
6
7
8
9
10
11
|
def send(messages : KeyedMessage[K,V]*) { lock synchronized { if (hasShutdown.get) //如果producer已经关闭了抛出异常退出 throw new ProducerClosedException recordStats(messages //更新producer统计信息 sync match { case true = > eventHandler.handle(messages) //如果是同步发送,直接使用DefaultEventHandler的handle方法发送 case false = > asyncSend(messages) // 否则,使用ayncSend方法异步发送消息——本文不考虑这种情况 } } } |
由上面的分析可以看出,真正的发送逻辑其实是由DefaultEventHandler类的handle方法来完成的。下面我们重点分析一下这个类的代码结构。
五、DefaultEventHandler与消息发送
这个类的handler方法可以同时支持同步和异步的消息发送。我们这里只考虑同步的代码路径。下面是消息发送的完整流程图:
以下代码是发送消息的核心逻辑:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
while (remainingRetries > 0 && outstandingProduceRequests.size > 0 ) { // 属性message.send.max.retries指定了消息发送的重试次数,而outstandingProducerRequests就是序列化之后待发送的消息集合 topicMetadataToRefresh ++ = outstandingProduceRequests.map( _ .topic) //将待发送消息所属topic加入到待刷新元数据的topic集合 if (topicMetadataRefreshInterval > = 0 && SystemTime.milliseconds - lastTopicMetadataRefreshTime > topicMetadataRefreshInterval) { //查看是否已过刷新元数据时间间隔 Utils.swallowError(brokerPartitionInfo.updateInfo(topicMetadataToRefresh.toSet, correlationId.getAndIncrement)) // 更新topic元数据信息 sendPartitionPerTopicCache.clear() //如果消息key是空,代码随机选择一个分区并记住该分区,以后该topic的消息都会往这个分区里面发送。sendPartitionPerTopicCache就是这个缓存 topicMetadataToRefresh.clear //清空待刷新topic集合 lastTopicMetadataRefreshTime = SystemTime.milliseconds } outstandingProduceRequests = dispatchSerializedData(outstandingProduceRequests) // 真正的消息发送方法 if (outstandingProduceRequests.size > 0 ) { // 如果还有未发送成功的消息 info( "Back off for %d ms before retrying send. Remaining retries = %d" .format(config.retryBackoffMs, remainingRetries- 1 )) // back off and update the topic metadata cache before attempting another send operation Thread.sleep(config.retryBackoffMs) // 等待一段时间并重试 // get topics of the outstanding produce requests and refresh metadata for those Utils.swallowError(brokerPartitionInfo.updateInfo(outstandingProduceRequests.map( _ .topic).toSet, correlationId.getAndIncrement)) sendPartitionPerTopicCache.clear() remainingRetries - = 1 // 更新剩余重试次数 producerStats.resendRate.mark() } } |
下面具体说说各个子模块的代码逻辑:
1
2
3
4
5
6
|
serializedMessages + = new KeyedMessage[K,Message]( topic = e.topic, key = e.key, partKey = e.partKey, message = new Message(bytes = encoder.toBytes(e.message))) // new Message时没有指定key |
构建完KeyedMessage之后返回对应的消息集合即可。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
def updateInfo(topics : Set[String], correlationId : Int) { var topicsMetadata : Seq[TopicMetadata] = Nil // TopicMetadata = topic信息+ 一组PartitionMetadata (partitionId + leader + AR + ISR) val topicMetadataResponse = ClientUtils.fetchTopicMetadata(topics, brokers, producerConfig, correlationId) //构造TopicMetadataRequest并随机排列所有broker,然后从第一个broker开始尝试发送请求。一旦成功就终止后面的请求发送尝试。 topicsMetadata = topicMetadataResponse.topicsMetadata //从response中取出zookeeper中保存的对应topic元数据信息 // throw partition specific exception topicsMetadata.foreach(tmd = >{ trace( "Metadata for topic %s is %s" .format(tmd.topic, tmd)) if (tmd.errorCode == ErrorMapping.NoError) { topicPartitionInfo.put(tmd.topic, tmd) //更新到broker的topic元数据缓存中 } else warn( "Error while fetching metadata [%s] for topic [%s]: %s " .format(tmd, tmd.topic, ErrorMapping.exceptionFor(tmd.errorCode).getClass)) tmd.partitionsMetadata.foreach(pmd = >{ if (pmd.errorCode ! = ErrorMapping.NoError && pmd.errorCode == ErrorMapping.LeaderNotAvailableCode) { warn( "Error while fetching metadata %s for topic partition [%s,%d]: [%s]" .format(pmd, tmd.topic, pmd.partitionId, ErrorMapping.exceptionFor(pmd.errorCode).getClass)) } // any other error code (e.g. ReplicaNotAvailable) can be ignored since the producer does not need to access the replica and isr metadata }) }) producerPool.updateProducer(topicsMetadata) } |
关于上面代码中的最后一行, 我们需要着重说一下。每个producer应用程序都会保存一个producer池对象来缓存每个broker上对应的同步producer实例。具体格式为brokerId -> SyncProducer。SyncProducer表示一个同步producer,其主要的方法是send,支持两种请求的发送:ProducerRequest和TopicMetadataRequest。前者是发送消息的请求,后者是更新topic元数据信息的请求。为什么需要这份缓存呢?我们知道,每个topic分区都应该有一个leader副本在某个broker上,而只有leader副本才能接收客户端发来的读写消息请求。对producer而言,即只有这个leader副本所在的broker才能接收ProducerRequest请求。在发送消息时候,我们会首先找出这个消息要发给哪个topic,然后发送更新topic元数据请求给任意broker去获取最新的元数据信息——这部分信息中比较重要的就是要获取topic各个分区的leader副本都在哪些broker上,这样我们稍后会创建连接那些broker的阻塞通道(blocking channel)去实现真正的消息发送。Kafka目前的做法就是重建所有topic分区的leader副本所属broker上对应的SyncProducer实例——虽然我觉得这样实现有线没有必要,只更新消息所属分区的缓存信息应该就够了(当然,这只是我的观点,如果有不同意见欢迎拍砖)。以下是更新producer缓存的一些关键代码:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
val newBrokers = new collection.mutable.HashSet[Broker] topicMetadata.foreach(tmd = > { tmd.partitionsMetadata.foreach(pmd = > { if (pmd.leader.isDefined) //遍历topic元数据信息中的每个分区元数据实例,如果存在leader副本的,添加到newBrokers中以备后面更新缓存使用 newBrokers+ = (pmd.leader.get) }) }) lock synchronized { newBrokers.foreach(b = > { //遍历newBrokers中的每个broker实例,如果在缓存中已经存在,直接关闭掉然后创建一个新的加入到缓存中;否则直接创建一个加入 if (syncProducers.contains(b.id)){ syncProducers(b.id).close() syncProducers.put(b.id, ProducerPool.createSyncProducer(config, b)) } else syncProducers.put(b.id, ProducerPool.createSyncProducer(config, b)) }) } |
前面说了,如果只发送一条消息的话,其实真正需要更新的分区leader副本所述broker对应的SyncProducer实例只有一个,但目前的代码中会更新所有分区,不知道Java版本的producer是否也是这样实现,这需要后面继续调研!
Topic | 分区 | Leader副本所在的broker ID |
test-topic | P0 | 0 |
test-topic | P1 | 1 |
test-topic | P2 | 3 |
如果基于这样的配置,假定我们使用producer API一次性发送4条消息,分别是M1,M2, M3和M4。现在就可以开始分析代码了,首先从消息分组及整理开始:
消息 | 要被发送到的分区ID | 该分区leader副本所在broker ID |
M1 | P0 | 0 |
M2 | P0 | 0 |
M3 | P1 | 1 |
M4 | P2 | 3 |
1
2
3
|
val index = Utils.abs(Random.nextInt) % availablePartitions.size // 随机确定broker id val partitionId = availablePartitions(index).partitionId sendPartitionPerTopicCache.put(topic, partitionId) // 加入缓存中以便后续使用 |
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
def startup() { ... // 创建一个请求处理的线程池,在构造时就会开启多个线程准备接收请求 requestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.requestChannel, apis, config.numIoThreads) ... } class KafkaRequestHandlerPool { ... for (i <- 0 until numThreads) { runnables(i) = new KafkaRequestHandler(i, brokerId, aggregateIdleMeter, numThreads, requestChannel, apis) threads(i) = Utils.daemonThread( "kafka-request-handler-" + i, runnables(i)) threads(i).start() // 启动每个请求处理线程 } ... } |
KafkaRequestHandler实际上是一个Runnable,它的run核心方法中以while (true)的方式调用api.handle(request)不断地接收请求处理,如下面的代码所示:
1
2
3
4
5
6
7
8
9
10
11
12
|
class KafkaRequestHandler... extends Runnable { ... def run() { ... while ( true ) { ... apis.handle(request) // 调用apis.handle等待请求处理 } ... } ... } |
在KafkaApis中handle的主要作用就是接收各种类型的请求。本文只关注ProducerRequest请求:
1
2
3
4
5
6
7
8
|
def handle(request : RequestChannel.Request) { ... request.requestId match { case RequestKeys.ProduceKey = > handleProducerOrOffsetCommitRequest(request) // 如果接收到ProducerRequest交由handleProducerOrOffsetCommitRequest处理 case ... } ... } |
如此看来,核心的方法就是handleProducerOrOffsetCommitRequest了。这个方法之所以叫这个名字,是因为它同时可以处理ProducerRequest和OffsetCommitRequest两种请求,后者其实也是一种特殊的ProducerRequest。从Kafka 0.8.2之后kafka使用一个特殊的topic来保存提交位移(commit offset)。这个topic名字是__consumer_offsets。本文中我们关注的是真正的ProducerRequest。下面来看看这个方法的逻辑,如下图所示:
整体逻辑看上去非常简单,如下面的代码所示:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
|
def handleProducerOrOffsetCommitRequest(request : RequestChannel.Request) { ... val localProduceResults = appendToLocalLog(produceRequest, offsetCommitRequestOpt.nonEmpty) // 将消息追加写入本地提交日志 val numPartitionsInError = localProduceResults.count( _ .error.isDefined) // 计算是否存在发送失败的分区 if (produceRequest.requiredAcks == 0 ) { // request.required.acks = 0时的代码路径 if (numPartitionsInError ! = 0 ) { info(( "Send the close connection response due to error handling produce request " + "[clientId = %s, correlationId = %s, topicAndPartition = %s] with Ack=0" ) .format(produceRequest.clientId, produceRequest.correlationId, produceRequest.topicPartitionMessageSizeMap.keySet.mkString( "," ))) requestChannel.closeConnection(request.processor, request) // 关闭底层Socket以告知客户端程序有发送失败的情况 } else { ... } } else if (produceRequest.requiredAcks == 1 || // request.required.acks = 0时的代码路径,当然还有其他两个条件 produceRequest.numPartitions < = 0 || numPartitionsInError == produceRequest.numPartitions) { val response = offsetCommitRequestOpt.map( _ .responseFor(firstErrorCode, config.offsetMetadataMaxSize)) .getOrElse(ProducerResponse(produceRequest.correlationId, statuses)) requestChannel.sendResponse( new RequestChannel.Response(request, new BoundedByteBufferSend(response))) // 发送response给客户端 |