Kafka的Producer实现原理剖析

剖析producer之前，我们来回顾一下Kafka的producer，producer（生产者）：消息放到队列里面的叫生产者。

producer的主要功能就是向某个topic的某个分区发送一条消息。每个producer都是独立工作的，与其他producer实例之间没有关联。为了实现producer往某个分区发送一条消息，所以producer需要确定到底给哪个分区写入消息--这是producer分区器（partitioner）需要做的事情。Kafka Producer提供一个默认的分区器，对于每一条待发送的消息而言，如果该消息指定了key，那么该 partitioner会根据key的哈希值来选择目标分区；若这条消息没有指定key，则partitioner使用轮询的方式确认目标分区——这样可以最大限度地确保消息在所有分区上的均匀性。

Kafka Producer的设计的工作原理如图：

producer首先使用一个线程（用户主线程，也就是用户启动producer的线程）将待发送的消息封装进一个 ProducerRecord 类实例，然后将其序列化之后发送给 partitioner，再由后者确定了目标分区后一同发送到位于 producer程序中的一块内存缓冲区中。而 producer的另一个工作线程（I/O发送线程，也称 Sender线程）则负责实时地从该缓冲区中提取出准备就绪的消息封装进一个批次（batch），统一发送给对应的broker。整个producer的工作流程大概就是这样的。

producer主要参数

bootstrap.servers

该参数指定了一组 host：port 对，用于创建向 Kafka broker 服务器的连接，比如 k1：9092，k2：9092，k3：9092。

key.serializer

该参数就是为消息的 key 做序列化之用的。有默认值。

value.serializer

和 key.serializer 类似，只是它被用来对消息体（即消息 value）部分做序列化，将消息value 部分转换成字节数组。

acks

acks参数用于控制 producer生产消息的持久性（durability）。

1. acks=0：设置成 0 表示 producer 完全不理睬 leader broker 端的处理结果。
2. acks=all或者-1：表示当发送消息时，leader broker不仅会将消息写入本地日志，同时还会等待 ISR 中所有其他副本都成功写入它们各自的本地日志后，才发送响应结果给producer。
3. acks=1：是 0 和 all 折中的方案，也是默认的参数值。producer 发送消息后 leader broker 仅将该消息写入本地日志，然后便发送响应结果给 producer，而无须等待 ISR中其他副本写入该消息。那么此时只要该leader broker一直存活，Kafka就能够保证这条消息不丢失。这实际上是一种折中方案，既可以达到适当的消息持久性，同时也保证了producer端的吞吐量。

buffer.memory

该参数指定了 producer 端用于缓存消息的缓冲区大小，单位是字节，默认值是 33554432，即 32MB。

compression.type

compression.type 参数设置 producer 端是否压缩消息，默认值是 none，即不压缩消息。

retries

Kafka broker 在处理写入请求时可能因为瞬时的故障（比如瞬时的 leader 选举或者网络抖动）导致消息发送失败。该参数表示进行重试的次数，默认值是 0，表示不进行重试。

batch.size

batch.size 是 producer 最重要的参数之一！它对于调优 producer 吞吐量和延时性能指标都有着非常重要的作用。batch.size 参数默认值是 16384，即 16KB。

linger.ms

linger.ms 参数就是控制消息发送延时行为的。该参数默认值是 0，表示消息需要被立即发送，无须关心 batch 是否已被填满，大多数情况下这是合理的。

max.request.size

该参数用于控制producer发送请求的大小。实际上该参数控制的是producer 端能够发送的最大消息大小。

request.timeout.ms

当 producer 发送请求给 broker 后，broker 需要在规定的时间范围内将处理结果返还给producer。这段时间便是由该参数控制的，默认是 30秒。

总结：

1：producer的主要功能就是向某个topic的某个分区发送一条消息。

2：ack设置是producer的核心参数。