apache kafka系列之Producer处理逻辑

 最近研究producer的负载均衡策略,,,,我在librdkafka里边用代码实现了partition 值的轮询方法,,,但是在现场验证时,他的负载均衡不起作用,,,所以来找找原因;
下文是一篇描写kafka处理逻辑的文章,转载过来,研究一下.
apache kafka系列之Producer处理逻辑 分类:
apache kafka(73) apache kafka系列之Producer处理逻辑

版权声明:本文为博主原创文章,未经博主允许不得转载。

 

目录(?)[+]

 

apache kafka中国社区QQ群:162272557

转载自同事(董重)写得一篇wiki博客

Kafka Producer处理逻辑

Kafka Producer产生数据发送给Kafka Server,具体的分发逻辑及负载均衡逻辑,全部由producer维护。

Kafka结构图

apache kafka系列之Producer处理逻辑

Kafka Producer默认调用逻辑

apache kafka系列之Producer处理逻辑

默认Partition逻辑

1、没有key时的分发逻辑

每隔 topic.metadata.refresh.interval.ms 的时间,随机选择一个partition。这个时间窗口内的所有记录发送到这个partition。

发送数据出错后也会重新选择一个partition

2、根据key分发

对key求hash,然后对partition数量求模

Utils.abs(key.hashCode) % numPartitions

如何获取Partition的leader信息(元数据)

决定好发送到哪个Partition后,需要明确该Partition的leader是哪台broker才能决定发送到哪里。

具体实现位置

kafka.client.ClientUtils#fetchTopicMetadata

 实现方案

1、从broker获取Partition的元数据。由于Kafka所有broker存有所有的元数据,所以任何一个broker都可以返回所有的元数据

2、broker选取策略:将broker列表随机排序,从首个broker开始访问,如果出错,访问下一个

3、出错处理:出错后向下一个broker请求元数据

注意

  • Producer是从broker获取元数据的,并不关心zookeeper。
  • broker发生变化后,producer获取元数据的功能不能动态变化。
  • 获取元数据时使用的broker列表由producer的配置中的 metadata.broker.list 决定。该列表中的机器只要有一台正常服务,producer就能获取元数据。
  • 获取元数据后,producer可以写数据到非 metadata.broker.list 列表中的broker

错误处理

producer的send函数默认没有返回值。出错处理有EventHandler实现。

DefaultEventHandler的错误处理如下:

  • 获取出错的数据
  • 等待一个间隔时间,由配置 retry.backoff.ms 决定这段时间长短
  • 重新获取元数据
  • 重新发送数据

出错重试次数由配置 message.send.max.retries 决定

所有重试全部失败时,DefaultEventHandler会抛出异常。代码如下

if(outstandingProduceRequests.size >0) {

producerStats.failedSendRate.mark()

val correlationIdEnd = correlationId.get()

error("Failed to send requests for topics %s with correlation ids in [%d,%d]"

.format(outstandingProduceRequests.map(_.topic).toSet.mkString(","),

correlationIdStart, correlationIdEnd-1))

thrownewFailedToSendMessageException("Failed to send messages after "+ config.messageSendMaxRetries +" tries.", null)

}

请注明转载自:http://write.blog.csdn.NET/postedit/26687109

上一篇:Kafka系列之-Kafka监控工具KafkaOffsetMonitor配置及使用


下一篇:java中的基本数据类型和引用数据类型