kafka集群Producer基本数据结构及工作流程深入剖析-kafka 商业环境实战

栏目: 后端 · 发布时间: 5年前

内容简介:本套技术专栏是作者(秦凯新)平时工作的总结和升华,通过从真实商业环境抽取案例进行总结和分享,并给出商业应用的调优建议和集群环境容量规划等内容,请持续关注本套博客。期待加入IOT时代最具战斗力的团队。QQ邮箱地址:1120746959@qq.com,如有任何学术交流,可随时联系。

本套技术专栏是作者(秦凯新)平时工作的总结和升华,通过从真实商业环境抽取案例进行总结和分享,并给出商业应用的调优建议和集群环境容量规划等内容,请持续关注本套博客。期待加入IOT时代最具战斗力的团队。QQ邮箱地址:1120746959@qq.com,如有任何学术交流,可随时联系。

1 Producer端基本数据结构

  • ProducerRecord: 一个ProducerRecord表示一条待发送的消息记录,主要由5个字段构成:

    topic          所属topic
      partition      所属分区
      key            键值
      value          消息体
      timestamp      时间戳
    复制代码
  • RecordMetadata: Kafka服务器端返回给客户端的消息的元数据信息,前3项相对比较重要,Producer端可以使用这些消息做一些消息发送成功之后的处理。

    offset                   该条消息的位移
      timestamp                消息时间戳
      topic + partition        所属topic的分区
      checksum                 消息CRC32码
      serializedKeySize        序列化后的消息键字节数
      serializedValueSize      序列化后的消息体字节数
    复制代码

2 Producer端消息发送流程

kafka集群Producer基本数据结构及工作流程深入剖析-kafka 商业环境实战
  • 在send()的发送消息动作触发之前,通过props属性中指定的servers连接到broker集群,从Zookeeper收集集群Metedata信息,从而了解哪些broker掌管哪一个Topic的哪一个partition,以及brokers的健康状态。

  • 下面就是流水线操作,ProducerRecord对象携带者topic,partition,message等信息,在Serializer这个“车间”被序列化。

  • 序列化过后的ProducerRecord对象进入Partitioner“车间”,按照上文所述的Partitioning 策略决定这个消息将被分配到哪个Partition中。

  • 确定partition的ProducerRecord进入一个缓冲区,通过减少IO来提升性能,在这个“车间”,消息被按照TopicPartition信息进行归类整理,相同Topic且相同parition的ProducerRecord被放在同一个RecordBatch中,等待被发送。什么时候发送?都在Producer的props中被指定了,有默认值,显然我们可以自己指定。

    (1) batch.size:设置每个RecordBatch可以缓存的最大字节数 
      (2) buffer.memory:设置所有RecordBatch的总共最大字节数 
      (3) linger.ms设置每个RecordBatch的最长延迟发送时间 
      (4) max.block.ms 设置每个RecordBatch的最长阻塞时间 
    复制代码
  • 一旦,当单个RecordBatch的linger.ms延迟到达或者batch.size达到上限,这个 RecordBatch会被立即发送。另外,如果所有RecordBatch作为一个整体,达到了buffer.memroy或者max.block.ms上限,所有的RecordBatch都会被发送。

  • ProducerRecord消息按照分配好的Partition发送到具体的broker中,broker接收保存消息,更新Metadata信息,同步给Zookeeper。

  • Producer端其他优化点:

    (5) acks:Producer的数据确认阻塞设置,0表示不管任何响应,只管发,发完了立即执行下个任务,这种方式最快,但是很不保险。1表示只确保leader成功响应,接收到数据。2表示确保leader及其所有follwer成功接收保存消息,也可以用”all”。
      (6) retries:消息发送失败重试的次数。
      (7) retry.backoff.ms:失败补偿时间,每次失败重试的时间间隔,不可设置太短,避免第一条消息的响应还没返回,第二条消息又发出去了,造成逻辑错误。
      (8) max.in.flight.request.per.connection:同一时间,每个Producer能够发送的消息上限。
      (9) compression.type  producer所使用的压缩器,目前支持gzip, snappy和lz4。压缩是在用户主线程完成的,通常都需要花费大量的CPU时间,但对于减少网络IO来说确实利器。生产环境中可以结合压力测试进行适当配置
    复制代码

以上就是本文的全部内容,希望本文的内容对大家的学习或者工作能带来一定的帮助,也希望大家多多支持 码农网

查看所有标签

猜你喜欢:

本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们

BSD Hacks

BSD Hacks

Dru Lavigne / O'Reilly Media, Inc. / 2004-05-24 / USD 24.95

If you want more than your average BSD user--you want to explore and experiment, unearth shortcuts, create useful tools, and come up with fun things to try on your own--BSD Hacks is a must-have. This ......一起来看看 《BSD Hacks》 这本书的介绍吧!

JSON 在线解析
JSON 在线解析

在线 JSON 格式化工具

XML 在线格式化
XML 在线格式化

在线 XML 格式化压缩工具

Markdown 在线编辑器
Markdown 在线编辑器

Markdown 在线编辑器