本章内容将整理下Kafka体系结构中的生产者相关的一些知识。

1. 生产者客户端

• 生产者客户端在Kafka的发展历程当中一共有两个重大版本：

  • 一个是基于Scala语言开发的版本，称为Old Producer或Scala版的生产者客户端。
  • 一个是Kafka0.9.x版本之后以Java语言开发的版本，称为New Producer或Java版的生产者客户端。

• 生产者客户端的3个必要参数配置：

  • bootstrap.servers：

    用来指定生产者客户端连接Kafka集群中的broker清单吗，格式为：host1:port1,host2:port2, …。

    没有必要将Kafka集群中的所有broker都列入到清单中，因为生产者客户端可以通过一个broker查询到其他broker。

    虽然清单中可以只配置一个broker，但是为了以防该broker宕机后导致生产者客户端无法正常发送消息，所以建议至少配置2个或2个以上的broker。

  • key.serializer 和 value.serializer：

    broker只接收字节数组（Byte[]）形式的消息。所以，发送消息的时候需要将消息的key和value都转换为字节数组。而key.serializer 和 value.serializer就是分别指定了key和value的序列化器。

• 消息发送

  • 消息发送有三种模式：

    • 发完即忘（fire-and-forget）

      只管往Kafka发送消息但是不管消息是否正确到达。

    • 同步（sync）

    • 异步（async）

      这里觉得值得说的是异步回调函数的调用循序问题，如果两个消息R1、R2都是发给了同一个分区，且R1先于R2，那么生产者客户端就能保证R1对应的回调函数优先于R2对应的回调函数被调用，也就是说回调函数的调用可以保证分区有序性。

  • 消息通过send()方法发送到broker的过程中，还需要经过拦截器、序列化器、分区器的操作：

    • 拦截器

      拦截器按着生产者客户端和消费者客户端分为：生产者拦截器和消费者拦截器。

      生产者拦截器可以使得我们有能力在消息真正发送出去之前做一些额外处理，对于消息来说生产者拦截器不是必需的。

    • 序列化器

      上文中讲道key.serializer和value.serializer的时候，有提到序列化器。这里提到是生产者客户端的序列化器，它将消息转换为字节数组以便后续通过网络传输给Kafka。

      与生产者客户端序列化器对应的是消费者客户端的反序列化器，即将字节数组转换为消息对象。

      生产者的序列化器与对应的消费者的反序列化器必须保持一一对应加粗样式，否则无法正确的解析消息。

      对于消息来说序列化器是必需的。

    • 分区器

      分区器的作用是给消息分配分区。

      默认分区器在key为null的情况下，会轮询的方式将消息发送给主题下各个可用的分区。如果key不为null，则会根据key来计算出该消息将要发往的分区的分区号，该分区可以是主题下的任意分区。

      分区器不是每次都是必需的。需要根据实际的情况来判断，如果已指定了消息的分区号（即消息ProducerRecord中指定了partition字段，partition代表的就是所要发往的分区的分区号），那么就不需要分区器来为消息分配分区了。

2. 原理分析

首先我们先来看一张生产者客户端的整体架构图：

从整体架构图中我们可以得知以下一些信息：

1. 生产者客户端是通过两个线程来协调工作的：主线程和Sender线程。

   • 主线程：用来创建消息，并使消息经过拦截器、序列化器、分区器的处理后，存储到消息累加器（RecordAccumulator，也称为消息收集器）中。
   • Sender线程：用来从消息累加器中获取消息并将消息发送给Kafka。

2. 消息累加器的空间不是无限的，我们可以通过生产者客户端的配置参数buffer.memory来进行配置，默认情况下是32M。

3. 消息写入消息累加器的速度如果远远超过消息发送到Kafka的速度，就会造成消息积压，从而导致消息累加器的空间被占满，即出现生产者客户端空间不足的情况。此时KafkaProducer的send()方法要么阻塞，要么抛出异常，这取决于生产者客户端的一个配置参数max.block.ms（最大阻塞超时时间），默认为60秒，即先阻塞，当阻塞时间超过了max.block.ms则抛异常。

4. 双端队列（Deque）：在消息累加器中有1个或多个双端队列，每个双端队列对应一个分区，双端队列中有若干ProducerBatch，ProducerBatch中存放的是若干个ProducerRecord。

   我们可以将ProducerRecord理解为一条消息，而ProducerBatch则是一个消息批次。

   消息被写入消息累加器的时候，会根据消息所属的分区找到对应的双端队列，然后取出该双端队列队尾的ProducerBatch并将消息写入到这里。

5. Sender线程是从双端队列的对头开始获取消息的。

6. BufferPool：消息在发送给Kafka之前需要一块内存空间来进行存储，我们姑且叫消息内存。生产者客户端是通过java.io.ByteBuffer类来对消息内存进行创建和释放的。

   消息内存频繁的创建和释放也是比较消耗新能的，为了减少这部分开销，消息累加器中提供BufferPool，在BufferPool中缓存了指定大小的ByteBuffer，这些ByteBuffer是可以进行复用（免去了再创建和释放同等大小的ByteBuffer开销）。

   注意这里提到的是“指定大小”，例如：只允许16KB大小的ByteBuffer能缓存在BufferPool中，那么大小不是16KB的ByteBuffer就不允许进入到BufferPool中，也就无法进行的复用了。

   我们可以通过生产者客户端的参数batch.size（默认16KB）来设置这个“指定大小”。

7. 生产者客户端的参数batch.size对ProducerBatch的大小也有很大的影响。当消息要写入消息累加器的时候，首先根据消息的分区获取对应的双端队列（如果没有则新建），然后获取处于双端队列队尾的ProducerBatch（如果没有则新建），尝试将消息写入到该ProducerBatch中，如果能写入则写入（即ProducerBatch空间还够），如果不能，则需要新建一个ProducerBatch。当我们要新建ProducerBatch的时候，首先比较下消息的大小是否超过batch.size，如果没超过，则基于batch.size大小创建一个新的ProducerBatch（此内存空间将会被复用，即被BufferPool缓存了）；如果超过了，则基于消息的大小去创建一个新的ProducerBatch（此内存空间将不会被复用，即没有缓存到BufferPool中）。

8. 基于以上出现的一些概念，分区、双端队列、ProducerBatch和ProducrRecord，我整理了一下它们之间的关系：
   

9. 消息写入到消息累加器的过程如下：
   

上一篇：Kafka之基本概念
下一篇：Kafka之消费者