Kafka 服务中生产者创建详解

2020-05-20 0 By admin

简单地说:生产者就是负责向 Kafka 发送消息的应用程序;但是其中包含了很多的操作。

一、生产者客户端开发

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

  1. 配置生产者客户端参数及创建相应的生产者实例。
  2. 构建待发送的消息。
  3. 发送消息。
  4. 关闭生产者实例。

1.1、创建生产者实例

在创建真正的生产者实例前需要配置项相应的参数。
1、bootstrap.servers : 改参数用来指定生产者客户端连接 Kafka 集群所需要的 broker 地址清单。不需要设置所有的 broker,因为生产者会从给定的 broker 中查找到其他的 broker 的信息。
2、key.serializer 和 value.serializer : broker 接受到的消息必须以字节数组(byte[])形式存在,使代码具有良好的可读性;不过在发往 broker 之前需要将消息中对应的 key 和 value 做相应的序列化操作来转换成字节数组。而这两个参数用来分别指定 key 和 value 序列化操作的序列化器。
另外还有其他各种参数,可以在创建生产者实例时设置。
1、client.id:这个参数用来设定KafkaProducer 对应的客户端id ,默认值为“” 。如果客户端不设置, 则KafkaProducer 会自动生成一个非空字符串,内容形式如“ producer-1”“producer-2”。
2、buffer.memory:生产者客户端中用于缓存消息的缓冲区大小,默认值为33554432(32MB)。
3、batch.size:用于指定ProducerBatch 可以复用内存区域的大小,默认值为16384(16KB)。
4、max.in.flight.requests.per.connection:限制每个连接(也就是客户端与Node之间的连接)最多缓存的请求数,默认值为5。
5、acks:指定分区中必须要有多少个副本收到这条消息,之后生产者才会认为这条消息是成功写入的。

1.2、构建消息对象

构建的消息对象 ProducerRecord ,并不是单纯的消息,它包含了多个属性。

  1. topic : 消息要发往的主题。(必填)
  2. partition :消息要发往的分区号。
  3. headers : 消息的头部 ,设置一些与应用相关的信息,可以省略。
  4. key : 指定消息的键。不仅是消息的附加信息,会用来计算消息发往的分区;同一个 key 的消息会划分到同一个分区中。
  5. value : 消息体。(必填)
  6. timestamp : 时间戳。

1.3、发送消息

创建生产者实例和构建消息对象后,就开始设置发送消息了。发送消息主要有三种模式:发后即忘(fire-and-forget)、同步sync和异步async。
1、发后即忘,它只管往Kafka 中发送消息而并不关心消息是否正确到达。
2、同步发送的方式可靠性高,要么消息被发送成功,要么发生异常。不过同步发送的方式的性能会差很多,需要阻塞等待一条消息发送完之后才能发送下一条。
3、异步发送,使用Callback 的方式非常简洁明了,Kafka有响应时就会回调,要么发送成功,要么抛出异常。

1.4、序列化

生产者需要用序列化器(Serializer)把对象转换成字节数组才能通过网络发送给 Kafka。而消费者需要用反序列化器(Deserializer)把从 Kafka 中收到的字节数组转换成相应的对象。
序列化器有好多种,生产者使用的序列化器和消费者使用的反序列化器要一一对应。

1.5、生产者分区器

消息在通过 send() 方法发往 broker 的过程中,有可能需要经过拦截器(Interceptor)、序列化器和分区器(Partitioner)的一系列作用之后才能被真正地发往 broker。
拦截器一般不必需的;消息经过序列化后就需要确定它发往哪个分区;如果消息中指定了 partition 字段,就不需要分区器处理了;如果没有指定,则依赖分区器根据 key 这个字段计算 partition 值。

1.6、生产者拦截器

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

二、生产者原理分析

Kafka 消息队列生产者原理图
Kafka 消息队列生产者

2.1、两个线程

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

2.2、消息收集器

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

2.3、双端队列 ProducerBatch

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

2.4、消息缓冲池

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