Kafka生产者

一、必要的参数配置

1.1、bootstrap.server

该参数用来指定生产者客户端连接Kafka集群所需的broker地址清单，具体的内容格式为host1:port1,host2:port2,可以设置一个或多个地址，中间以逗号隔开，此参数的默认值为“”

1.2、key.serializer和value.serializer

broker端接收的消息必须以字节数组(byte[])的形式存在。在发往broker之前需要将消息中对应的key和value做相应的序列化操作来转换成字节数组。key.serializer和value.serializer这两个参数分别用来指定key和value序列化操作的序列化器，这两个参数无默认值。

1.3、ack

这个参数用来指定分区中必须要有多少个副本收到这条消息，之后生产者才会认为这条消息是成功写入的。acks是生产者客户端中一个非常重要的参数，它涉及消息的可靠性和吞吐量之间的权衡。acks参数有3种类型的值(都是字符串类型)。

1、acks=1。默认值即为1。生产者发送消息之后，只要分区的leader副本成功写入消息，那么它就会收到来自服务端的成功响应。
如果消息无法写入leader副本，比如在leader副本崩溃、重新选举新的leader副本的过程中，那么生产者就会收到一个错误的响应，
为了避免消息丢失，生产者可以选择重发消息。如果消息写入leader副本并返回成功响应给生产者，
且在被其他follower副本拉取之前leader副本崩溃，那么此时消息还是会丢失，因为新选举的leader副本中并没有这条对应的消息。
acks设置为
1,是消息可靠性和吞吐量之间的折中方案。

2、acks=0。生产者发送消息之后不需要等待任何服务端的响应。
如果在消息从发送到写入Kafka的过程中出现某些异常，导致Kafka并没有收到这条消息，那么生产者也无从得知，消息也就丢失了。
在其他配置环境相同的情况下，acks设置为0可以达到最大的吞吐量。

3、acks=-1或acks=all。生产者在消息发送之后，需要等待ISR中的所有副本都成功写入消息之后才能够收到来自服务端的成功响应。
在其他配置环境相同的情况下，acks设置为-1(all)可以达到最强的可靠性。
但这并不意味着消息就一定可靠，因为ISR中可能只有leader副本，这样就退化成了acks=1的情况。
要获得更高的消息可靠性需要配合min.insync.replicas等参数的联动

1.4、max.request.size

这个参数用来限制生产者客户端能发送的消息的最大值，默认值为1048576B,即1MB。一般情况下，这个默认值就可以满足大多数的应用场景了。

1.5、retries和retry.backoff.ms

retries参数用来配置生产者重试的次数，默认值为0,即在发生异常的时候不进行任何重试动作。消息在从生产者发出到成功写入服务器之前可能发生一些临时性的异常，比如网络抖动、leader副本的选举等，这种异常往往是可以自行恢复的，生产者可以通过配置retries大于0的值，以此通过内部重试来恢复而不是一味地将异常抛给生产者的应用程序。如果重试达到设定的次数，那么生产者就会放弃重试并返回异常。不过并不是所有的异常都是可以通过重试来解决的，比如消息太大，超过max.request.size参数配置的值时，这种方式就不可行了。

重试还和另一个参数retry.backoff.ms有关，这个参数的默认值为100,它用来设定两次重试之间的时间间隔，避免无效的频繁重试。在配置retries和retry.backoff.ms之前，最好先估算一下可能的异常恢复时间，这样可以设定总的重试时间大于这个异常恢复时间，以此来避免生产者过早地放弃重试。

Kafka可以保证同一个分区中的消息是有序的。如果生产者按照一定的顺序发送消息，那么这些消息也会顺序地写入分区，进而消费者也可以按照同样的顺序消费它们。对于某些应用来说，顺序性非常重要，比如MySQL的binlog传输，如果出现错误就会造成非常严重的后果。如果将acks参数配置为非零值，并且max.in.flight.requests.per.connection参数配置为大于1的值，那么就会出现错序的现象：如果第一批次消息写入失败，而第二批次消息写入成功，那么生产者会重试发送第一批次的消息，此时如果第一批次的消息写入成功，那么这两个批次的消息就出现了错序。一般而言，在需要保证消息顺序的场合建议把参数max.in.flight.requests.per.connection配置为1,而不是把acks配置为0,不过这样也会影响整体的吞吐。

1.6、compression.type

这个参数用来指定消息的压缩方式，默认值为“none”,即默认情况下，消息不会被压缩。该参数还可以配置为“gzip”“snappy”和“Iz4”。对消息进行压缩可以极大地减少网络传输量、降低网络I/O,从而提高整体的性能。消息压缩是一种使用时间换空间的优化方式，如果对时延有一定的要求，则不推荐对消息进行压缩。

1.7、connections.max.idle.ms

这个参数用来指定在多久之后关闭限制的连接，默认值是540000(ms),即9分钟。

1.8、linger.ms

这个参数用来指定生产者发送ProducerBatch之前等待更多消息(ProducerRecord)加入ProducerBatch的时间，默认值为0。生产者客户端会在ProducerBatch被填满或等待时间超过linger.ms值时发送出去。增大这个参数的值会增加消息的延迟，但是同时能提升一定的吞吐量。这个linger.ms参数与TCP协议中的Nagle算法有异曲同工之妙。

1.9、receive.buffer.bytes

这个参数用来设置Socket接收消息缓冲区(SO_RECBUF)的大小，默认值为32768(B),即32KB。如果设置为-1,则使用操作系统的默认值。如果Producer与Kafka处于不同的机房，则可以适地调大这个参数值。

1.10、send.buffer.bytes

这个参数用来设置Socket发送消息缓冲区(SO_SNDBUF)的大小，默认值为131072(B),即128KB。与receive.buffer.bytes参数一样，如果设置为-1,则使用操作系统的默认值。

1.11、request.timeout.ms

这个参数用来配置Producer等待请求响应的最长时间，默认值为30000(ms)。请求超时之后可以选择进行重试。注意这个参数需要比broker端参数replica.lag.time.max.ms的值要大，这样可以减少因客户端重试而引起的消息重复的概率。

二、拦截器

Kafka一共有两种拦截器：生产者拦截器和消费者拦截器。

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

三、序列化

生产者需要用序列化器(Serializer)把对象转换成字节数组才能通过网络发送给Kafka。而在对侧，消费者需要用反序列化器(Deserializer)把从Kafka中收到的字节数组转换成相应的对象。生产者使用的序列化器和消费者使用的反序列化器是需要一一对应的，如果生产者使用了某种序列化器，比如StringSerializer,而消费者使用了另一种序列化器，比如IntegerSerializer,那么是无法解析出想要的数据的。

四、分区器

消息在通过sendO方法发往broker的过程中，有可能需要经过拦截器(Interceptor)、序列化器(Serializer)和分区器(Partitioner)的一系列作用之后才能被真正地发往broker。拦截器(下一章会详细介绍)一般不是必需的，而序列化器是必需的。消息经过序列化之后就需要确定它发往的分区，如果消息ProducerRecord中指定了partition字段，那么就不需要分区器的作用，因为partition代表的就是所要发往的分区号。

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

如果key不为null,那么默认的分区器会对key进行哈希(采用MurmurHash2算法，具备高运算性能及低碰撞率),最终根据得到的哈希值来计算分区号，拥有相同key的消息会被写入同一个分区。如果key为null,那么消息将会以轮询的方式发往主题内的各个可用分区。

五、生产者架构

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

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

在RecordAccumulator的内部还有一个BufferPool,它主要用来实现ByteBuffer的复用，以实现缓存的高效利用。不过BufferPool只针对特定大小的ByteBuffer进行管理，而其他大小的ByteBuffer不会缓存进BufferPool中，这个特定的大小由batch.size参数来指定，默认值为16384B,即16KB。我们可以适当地调大batch.size参数以便多缓存一些消息。

buffer.memory=32M
max.block.ms=60s
batch.size=16k