zoukankan      html  css  js  c++  java
  • 深入理解Kafka-学习笔记02-生产者整体架构

    生产者客户端的整体架构如下图所示:

      整个生产者客户端由两个线程协调运行,这两个线程分别为主线程和Sender线程(发送线程)。在主线程中由KafkaProducer创建消息,然后通过可能的拦截器、序列化器和分区器的作用之后缓存到消息累加器(RecordAccumulator,也称为消息收集器〉中。Sender线程负责从RecordAccumulator中获取消息并将其发送到Kafka中。

      RecordAccumulator主要用来缓存消息以便Sender线程可以批量发送,进而减少网络传输的资源消耗以提升性能。RecordAccumulator缓存的大小可以通过生产者客户端参数buffer.memory配置,默认值为33554432B,即32M。如果生产者发送消息的速度超过发送到服务器的速度,则会导致生产者空间不足,这个时候KafkaProducer的send()方法调用要么被阻塞,要么抛出异常,这个取决于参数max.block.ms的配置,此参数的默认值为60000,即60秒。

      RecordAccumulator的内部为每个分区都维护了一个双端队列,队列中的内容就是ProducerBatch,即Deque<ProducerBatch>。ProducerBatch中可以包含一至多个ProducerRecord。RecordAccumulator的内部还有一个BufferPool,它主要用来实现ByteBuffer的复用,以实现缓存的高效利用。不过BufferPool只针对特定大小的ByteBuffer进行管理,而其他大小的ByteBuffer不会缓存进BufferPool中,这个特定的大小由batch.size参数来指定,默认值为16384B,即16KB。

      ProducerBatch的大小和batch.size参数也有着密切的关系。当一条消息(ProducerRecord)流入RecordAccumulator时,会先寻找与消息分区所对应的双端队列(如果没有则新建),再从这个双端队列的尾部获取一个ProducerBatch(如果没有则新建),查看ProducerBatch中是否还可以写入这个ProducerRecord,如果可以则写入,如果不可以则需要创建一个新的ProducerBatch。在新建ProducerBatch时评估这条消息的大小是否超过batch.size参数的大小,如果不超过,那么就以batch.size参数的大小来创建ProducerBatch,这样在使用完这段内存区域之后,可以通过BufferPool的管理来进行复用;如果超过,那么就以评估的大小来创建ProducerBatch,这段内存区域不会被复用。生产者客户端会在 ProducerBatch 被填满或等待时间超过linger.ms 值时发送出去。

       请求在从Sender线程发往Kafka之前还会保存到InFlightRequests中,InFlightRequests保存对象的具体形式为Map<Nodeld,Deque<R巳quest>>,它的主要作用是缓存了已经发出去但还没有收到响应的请求(Nodeld是一个String类型,表示节点的id编号)。与此同时,InFlightRequests还提供了许多管理类的方法,并且通过配置参数还可以限制每个连接(也就是客户端与Node之间的连接)最多缓存的请求数。这个配置参数为max.工n.flight.requests.per.connection,默认值为5,即每个连接最多只能缓存5个未响应的请求,超过该数值之后就不能再向这个连接发送更多的请求了,除非有缓存的请求收到了响应(Response)。通过比较Deque<Request>的size与这个参数的大小来判断对应的Node中是否己经堆积了很多未响应的消息,如果真是如此,那么说明这个Node节点负载较大或网络连接有问题,再继续向其发送请求会增大请求超时的可能。

  • 相关阅读:
    Day3----《Pattern Recognition and Machine Learning》Christopher M. Bishop
    Day2----《Pattern Recognition and Machine Learning》Christopher M. Bishop
    学习笔记-----《Pattern Recognition and Machine Learning》Christopher M. Bishop
    “数学之美”笔记
    win10下使用nodejs安装及webstorm创建express项目的指导
    deepin/ubuntu下搭建Jekyll环境
    struts2.3.23升级到struts2.3.32
    struts2.5能不能再恶心点
    线程通信、线程同步以及进城通信的真的搞懂了么
    WINFORM数据库操作,有点像安装里面的SQLITE
  • 原文地址:https://www.cnblogs.com/java-meng/p/15210589.html
Copyright © 2011-2022 走看看