首页 > 代码库 > apache kafka源码分析走读-Producer分析
apache kafka源码分析走读-Producer分析
apache kafka中国社区QQ群:162272557
producer的发送方式剖析
Kafka提供了Producer类作为java producer的api,该类有sync和async两种发送方式。
sync架构图
async架构图
调用流程如下:
代码流程如下:
Producer:当new Producer(new ProducerConfig()),其底层实现,实际会产生两个核心类的实例:Producer、DefaultEventHandler。在创建的同时,会默认new一个ProducerPool,即我们每new一个java的Producer类,就会有创建Producer、EventHandler和ProducerPool,ProducerPool为连接不同kafka broker的池,初始连接个数有broker.list参数决定。
调用producer.send方法流程:
当应用程序调用producer.send方法时,其内部其实调的是eventhandler.handle(message)方法,eventHandler会首先序列化该消息,
eventHandler.serialize(events)-->dispatchSerializedData()-->partitionAndCollate()-->send()-->SyncProducer.send()
调用逻辑解释:当客户端应用程序调用producer发送消息messages时(既可以发送单条消息,也可以发送List多条消息),调用eventhandler.serialize首先序列化所有消息,序列化操作用户可以自定义实现Encoder接口,下一步调用partitionAndCollate根据topics的messages进行分组操作,messages分配给dataPerBroker(多个不同的Broker的Map),根据不同Broker调用不同的SyncProducer.send批量发送消息数据,SyncProducer包装了nio网络操作信息。
Producer的sync与async发送消息处理,大家看以上架构图一目了然。
partitionAndCollate方法详细作用:获取所有partitions的leader所在leaderBrokerId(就是在该partiionid的leader分布在哪个broker上),
创建一个HashMap<int, Map<TopicAndPartition, List<KeyedMessage<K,Message>>>>,把messages按照brokerId分组组装数据,然后为SyncProducer分别发送消息作准备工作。
名称解释:partKey:分区关键字,当客户端应用程序实现Partitioner接口时,传入参数key为分区关键字,根据key和numPartitions,返回分区(partitions)索引。记住partitions分区索引是从0开始的。
Producer平滑扩容机制
def handle(events: Seq[KeyedMessage[K,V]]) { ...... while (remainingRetries > 0 && outstandingProduceRequests.size > 0) { topicMetadataToRefresh ++= outstandingProduceRequests.map(_.topic) if (topicMetadataRefreshInterval >= 0 && SystemTime.milliseconds - lastTopicMetadataRefreshTime > topicMetadataRefreshInterval) { Utils.swallowError(brokerPartitionInfo.updateInfo(topicMetadataToRefresh.toSet, correlationId.getAndIncrement)) sendPartitionPerTopicCache.clear() topicMetadataToRefresh.clear 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)) //休眠时间,多长时间刷新一次 Thread.sleep(config.retryBackoffMs) // 生产者定期请求刷新最新topics的broker元数据信息 Utils.swallowError(brokerPartitionInfo.updateInfo(outstandingProduceRequests.map(_.topic).toSet, correlationId.getAndIncrement)) ..... } } }
BrokerPartitionInfo的updateInfo方法代码如下:
def updateInfo(topics: Set[String], correlationId: Int) { var topicsMetadata: Seq[TopicMetadata] = Nil //根据topics列表,meta.broker.list,其他配置参数,correlationId表示请求次数,一个计数器参数而已 //创建一个topicMetadataRequest,并随机的选取传入的broker信息中任何一个去取metadata,直到取到为止 val topicMetadataResponse = ClientUtils.fetchTopicMetadata(topics, brokers, producerConfig, correlationId) topicsMetadata = http://www.mamicode.com/topicMetadataResponse.topicsMetadata>
def fetchTopicMetadata(topics: Set[String], brokers: Seq[Broker], producerConfig: ProducerConfig, correlationId: Int): TopicMetadataResponse = { var fetchMetaDataSucceeded: Boolean = false var i: Int = 0 val topicMetadataRequest = new TopicMetadataRequest(TopicMetadataRequest.CurrentVersion, correlationId, producerConfig.clientId, topics.toSeq) var topicMetadataResponse: TopicMetadataResponse = null var t: Throwable = null val shuffledBrokers = Random.shuffle(brokers) //生成随机数 while(i < shuffledBrokers.size && !fetchMetaDataSucceeded) { //对随机选到的broker会创建一个SyncProducer val producer: SyncProducer = ProducerPool.createSyncProducer(producerConfig, shuffledBrokers(i)) info("Fetching metadata from broker %s with correlation id %d for %d topic(s) %s".format(shuffledBrokers(i), correlationId, topics.size, topics)) try { //发送topicMetadataRequest到该broker去取metadata,获得该topic所对应的所有的broker信息 topicMetadataResponse = producer.send(topicMetadataRequest) fetchMetaDataSucceeded = true } catch { ...... } } if(!fetchMetaDataSucceeded) { throw new KafkaException("fetching topic metadata for topics [%s] from broker [%s] failed".format(topics, shuffledBrokers), t) } else { debug("Successfully fetched metadata for %d topic(s) %s".format(topics.size, topics)) } return topicMetadataResponse }
def updateProducer(topicMetadata: Seq[TopicMetadata]) { val newBrokers = new collection.mutable.HashSet[Broker] topicMetadata.foreach(tmd => { tmd.partitionsMetadata.foreach(pmd => { if(pmd.leader.isDefined) newBrokers+=(pmd.leader.get) }) }) lock synchronized { newBrokers.foreach(b => { 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)) }) } }
当我们启动kafka broker后,并且大量producer和consumer时,经常会报如下异常信息。
root@lizhitao:/opt/soft$ Closing socket connection to 192.168.11.166
笔者也是经常很长时间看源码分析,才明白了为什么ProducerConfig配置信息里面并不要求使用者提供完整的kafka集群的broker信息,而是任选一个或几个即可。因为他会通过您选择的broker和topics信息而获取最新的所有的broker信息。
值得了解的是用于发送TopicMetadataRequest的SyncProducer虽然是用ProducerPool.createSyncProducer方法建出来的,但用完并不还回ProducerPool,而是直接Close.
这里有三点需要强调:
- 客户端调用send, 才会新建SyncProducer,只有调用send才会去定期刷新metadata
- 在每次取metadata时,kafka会新建一个SyncProducer去取metadata,逻辑处理完后再close。
- 根据当前SyncProducer(一个Broker的连接)取得的最新的完整的metadata,刷新ProducerPool中到broker的连接.
- 每10分钟的刷新会直接重新把到每个broker的socket连接重建,意味着在这之后的第一个请求会有几百毫秒的延迟。如果不想要该延迟,把topic.metadata.refresh.interval.ms值改为-1,这样只有在发送失败时,才会重新刷新。Kafka的集群中如果某个partition所在的broker挂了,可以检查错误后重启重新加入集群,手动做rebalance,producer的连接会再次断掉,直到rebalance完成,那么刷新后取到的连接着中就会有这个新加入的broker。
在ProducerPool中,SyncProducer的数目是由该topic的partition数目控制的,即每一个SyncProducer对应一个broker,内部封了一个到该broker的socket连接。每次刷新时,会把已存在SyncProducer给close掉,即关闭socket连接,然后新建SyncProducer,即新建socket连接,去覆盖老的。
如果不存在,则直接创建新的。