一个正常的生产者逻辑需要具备以下几个步骤:

  • 配置生产者客户端参数以及创建相应的生产者实例
  • 构建待发送的消息
  • 发送消息
  • 关闭生产者实例

序列化器

生产者在发送消息时,需要用序列化器(Serializer)把对象转换成字节数组才能通过网络发送给kafka,同样的消费者需要用反序列化器(Deserializer)把从Kafka中收到的字节数组转换成相应的对象。

生产者使用的序列化器和消费者使用的反序列化器需要一一对应。

kafka 客户端提供了多种序列化器,同时也支持自定义实现序列化器(可以选择使用如 Avro、JSON、Thrift、ProtoBuf、Protostuff等通用的序列化工具来实现)。

分区器

消息在通过 send() 方法发往 broke 的过程中,有可能需要经过拦截器(Interceptor)、序列化器(Serializer)和分区器(Partitioner)的一些列作用之后才能被真正的发往kafka。拦截器一般不是必须的,而序列化器是必需的。

消息经过序列化器之后就需要确定他发往的分区,如果消息 ProducerRecord 中指定了具体的 partition 字段,那么就不需要分区器的作用,因为partition 就代表了要发往的分区号。

如果消息没有指定分区号,那么就需要依赖分区器,根据 key 这个字段来计算 partition 的值。分区器的作用就是为消息分配分区。

kafka 中提供的默认分区器的计算逻辑是:

  • 如果 key 不为 null,则对 key 进行 哈希 (采用 MurmurHash2 算法),最终根据到的哈希值计算分区号,拥有相同key的消息会被发送到同一个分区
  • 如果 key 为 null,消息将会以轮训的方式发往主题内的各个可用分区

在不改变主题分区的数量下,key 和分区之间的映射关系可以保持不变,一旦主题的分区数量发生变化(增加分区),那么映射关系就很难保证了。

除了 kafka 默认提供的分区器外,还可以使用自定义的分区器,只需要同 DefaultPartitioner 一样实现 Partitioner 接口即可。

实现自定义分区器之后,需要通过配置参数显示指定自定义分区器。

拦截器

kafka 一共有两种拦截器:生产者拦截器和消费者拦截器。

生产者拦截器既可以在 消息发送前 做一些准备工作,比如按照某个规则过滤不符合条件的消息、修改消息的内容等,也可以在 发送回调逻辑前 做一些定制化的需求,比如统计类的工作。

实现方式:实现 ProducerInterceptor 接口,该接口提供如下三个方法

  • onSend
  • onAcknowledgement
  • close

KafkaProducer 再将消息序列化和计算分区之前会调用生产者拦截器的 onSend() 方法来对消息进行相应的定制化操作。(一般来说最好不要修改 ProducerRecord 的 topic key partition 等信息,不然有可能影响分区的计算以及broker端日志的压缩功能)

KafkaProducer 会在消息被应答之前和消息发送失败时调用生产者拦截器的 onAcknowledgement() 方法,优先于用户设定的 Callback 之前执行。这个方法运行在 Producer 的 IO 线程中,所以逻辑越简单越好,否则会影响消息发送的速度。

close() 方法主要用于在关闭拦截器时,执行一些资源的清理工作。

这三个方法中抛出的异常会被捕获并记录到日志中,并不会向上传递。

整体架构

kafka 中同个生产者客户端由两个线程协调运行,分别为:主线程和 Sender 线程(发送线程)。

在主线程中由 KafkaProducer 创建消息,然后通过 可能的 拦截器、序列化器、分区器的作用之后,缓存到消息累加器 RecordAccumulator 中。Sender 线程负责从消息累加器中获取消息并将其发送到 kafka 中。

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

主线程中发送过来的消息都会被追加到 RecordAccumulator 的某个双端队列(Deque)中,在 RecordAccumulator 的内部为每个分区都维护了一个双端队列,队列中的内容就是ProducerBatch ,即 Deque<ProducerBatch>。消息写入缓存时,追加到双端队列的尾部;Sender读取消息时,从双端队列的头部读取。注意ProducerBatch不是ProducerRecord,ProducerBatch中可以包含一至多个ProducerRecord。通俗地说,ProducerRecord 是生产者中创建的消息,而ProducerBatch是指一个消息批次,ProducerRecord会被包含在ProducerBatch中,这样可以使字节的使用更加紧凑。与此同时,将较小的ProducerRecord拼凑成一个较大的ProducerBatch,也可以减少网络请求的次数以提升整体的吞吐量。如果生产者客户端需要向很多分区发送消息,则可以将buffer.memory参数适当调大以增加整体的吞吐量。

消息在网络上都是以字节(Byte)的形式传输的,在发送之前需要创建一块内存区域来保存对应的消息。在Kafka生产者客户端中,通过java.io.ByteBuffer实现消息内存的创建和释放。不过频繁的创建和释放是比较耗费资源的,在RecordAccumulator的内部还有一个BufferPool,它主要用来实现ByteBuffer的复用,以实现缓存的高效利用。不过BufferPool只针对特定大小的ByteBuffer进行管理,而其他大小的ByteBuffer不会缓存进BufferPool中,这个特定的大小由batch.size参数来指定,默认值为16384B,即16KB。我们可以适当地调大batch.size参数以便多缓存一些消息。

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

Sender 从 RecordAccumulator 中获取缓存的消息之后,会进一步将原本<分区,Deque<ProducerBatch>>的保存形式转变成<Node,List<ProducerBatch>的形式,其中Node表示Kafka集群的broker节点。对于网络连接来说,生产者客户端是与具体的broker节点建立的连接,也就是向具体的broker 节点发送消息,而并不关心消息属于哪一个分区;而对于 KafkaProducer的应用逻辑而言,我们只关注向哪个分区中发送哪些消息,所以在这里需要做一个应用逻辑层面到网络I/O层面的转换。

在转换成<Node,List<ProducerBatch>>的形式之后,Sender 还会进一步封装成<Node,Request>的形式,这样就可以将Request请求发往各个Node了,这里的Request是指Kafka的各种协议请求,对于消息发送而言就是指具体的ProduceRequest

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

leastLoadedNode

InFlightRequests还可以获得leastLoadedNode,即所有Node中负载最小的那一个。这里的负载最小是通过每个Node在InFlightRequests中还未确认的请求决定的,未确认的请求越多则认为负载越大。

图中展示了三个节点Node0、Node1和Node2,很明显Node1的负载最小。也就是说,Node1为当前的leastLoadedNode。选择leastLoadedNode发送请求可以使它能够尽快发出,避免因网络拥塞等异常而影响整体的进度。leastLoadedNode的概念可以用于多个应用场合,比如元数据请求、消费者组播协议的交互。

当客户端中没有需要使用的元数据信息时,比如没有指定的主题信息,或者超过metadata.max.age.ms 时间没有更新元数据都会引起元数据的更新操作。客户端参数metadata.max.age.ms的默认值为300000,即5分钟。元数据的更新操作是在客户端内部进行的,对客户端的外部使用者不可见。当需要更新元数据时,会先挑选出leastLoadedNode,然后向这个Node发送MetadataRequest请求来获取具体的元数据信息。这个更新操作是由Sender线程发起的,在创建完MetadataRequest之后同样会存入InFlightRequests,之后的步骤就和发送消息时的类似。元数据虽然由Sender线程负责更新,但是主线程也需要读取这些信息,这里的数据同步通过synchronized和final关键字来保障。

评论