上篇学习总结了producer端的整体架构，producer的构造，分区器的相关知识点，以及序列化器的一些内容，对producer有一个初步的认知，本篇对kafka中的拦截器，消息压缩及安全，多线程等内容进行总结。
Producer中的拦截器是在kafka0.10.0.0中引进的一个新功能，用于实现clients端的定制化控制逻辑。拦截器可以帮助用户在消息发送前及回调逻辑前对消息做定制化需求，例如修改消息。另外，还可指定多个拦截器按序作用域同一条消息从而形成一个拦截链，拦截器的实现接口为org.apache.kafka.clients.producer.ProducerInterceptor。该接口的主要方法如下:
onSend(ProducerRecord):该方法被封装在KafkaProducer.send方法中，运行在用户主线程中，producer在消息被序列化之后，未计算分区前调用该方法，可以通过该方法对消息做任意操作，但是不建议修改消息所属的topic和分区，因为会影响目标分区的计算。
onAcknowledgement(RecordMetadata, Exception):该方法会在消息被应答之前或消息发送失败时调用，通常在producer回调逻辑触发之前，onAcknowlegement运行在producer的I/O线程中，不建议在该方法中放入重逻辑，因为会降低消息发送效率。
close:该方法用来关闭拦截器，主要是用于进行资源清理。
Kafka的消息发送采用异步发送机制，send方法将消息放入缓冲区，然后由专属I/O线程从缓冲区中提取消息封装进batch,然后发送出去，该过程存在数据丢失的风险。另外，消息发送过程中还可能出现乱序的风险。解决这两个问题，可以通过以下配置解决:
设置如下:
blocks.onbuffer.full=true:设置该参数为true时，当内存缓冲区被填满时producer处于阻塞状态并停止接收新的消息而不是抛出异常，否则producer生产速度过快会耗尽缓冲区。新版本0.10.0.0后使用max.block.ms参数代替。
acks=-1:设置所有ISR副本都响应了发送消息才能认为提交成功。
retries=Integer.MAX_VALUE:producer可以开启无限重试。
max.in.flight.requests.per.connection=1:该参数解决了topic同分区下的消息乱序问题，实际上限制了producer在单个broker连接上能够发送的未响应请求的数量，当设置为1时，producer在某个broker发送响应之前将无法再给该broker发送producer请求。
unclean.leader.election.enable=false:该参数关闭unclean leader选举，即不允许非ISR中的副本被选举为leader,避免造成消息丢失。
replication.factor=3:该参数设置副本数为3。
min.insync.replicas=2:该参数用于设置某条消息至少被写入到ISR的多少个副本中才算成功，只有acks参数为-1(all)时才生效。实际生产中要确保replication.factor>min.insync.replicas。
enable.auto.commit=false:禁止自动提交。
kafka中涉及大量数据读写，如何提高数据读写的性能呢，便是通过压缩实现，数据压缩能够显著降低磁盘占用或者带宽占用，从而有效地提升了I/O密集型应用的性能。但是压缩会消耗额外的CPU时钟周期，因此压缩是I/O性能和CPU资源的平衡。
kafka的压缩策略为:producer端压缩，broker端保持，consumer端解压。即producer端将一批消息压缩成一条消息发送，然后在broker端将这条压缩消息写入本地日志文件，通常broker端不会进行解压缩操作，只是原样保存消息，但如果某些前置条件不满足，例如需要进行消息格式转换，则需要在broker端进行解压再重新压缩。当consumer获取到这条压缩消息后，会自动对消息进行解压缩，还原成原来的消息集合。
在压缩算法方面，kafka支持3种压缩方法，GZIP，Snappy,LZ4,但是性能方面，LZ4>>Snappy>>GZIP,因为kafka源代码对Snappy中某个关键参数进行了硬编码。
在生产中，需要根据I/O资源消耗与CPU资源消耗对比来评估是否启用压缩。如果I/O资源紧张而CPU资源充足，则可以开启producer，反之不建议开启。
在kafka中，为了提高系统吞吐量，通常构造或者多个线程或多个进程来同时给Kafka集群发送消息，有两种方法:
1).多线程单KafkaProducer实例:这种方法是在全局构造一个kafkaProducer实例，然后在多个线程中共享使用，这种方法是线程安全的。实现简单，性能好，但是所有线程共享一个内存缓冲区，需要较多内存，另外，一旦producer某个线程崩溃导致KafkaProducer实例异常，则所有的用户线程都无法正常工作。
2).多线程多KafkaProducer实例:在每个Producer主线程中都构造一个KafkaProducer实例，并保证该实例在该线程中封闭，每个用户线程拥有专属的KafkaProducer实例，缓冲区空间及一组对应的配置参数，可以进行细粒度的调优，另外，不同KafkaProducer之间互不影响，即单个kafkaProducer的奔溃不会影响其他producer线程工作，但是需要较大的内存分配开销。对于分区数较少的kafka集群，建议使用单KafkaProducer，对于分区较多的集群，建议使用多KafkaProducer实例。
kafka0.9.0.0之前版本的producer被称为旧producer，0.9.0.0及之后的版本称为新producer，新旧版本producer主要有以下区别:
1).旧版本producer使用Scala语言编写，新版本使用Java编写。
2).旧版本producer的主要入口为kafka.producer.Producer,新版本producer的主要入口是org.apache.kafka.clients.producer.KafkaProducer。
3).旧版本producer发送方式默认为同步发送，新版本producer的发送方式默认为异步发送。
4)旧版本producer引入kafka-core依赖就可以直接使用，新版本需要引入kafka-clients依赖。
5).旧版本依赖Zookeeper发送数据，新版本不依赖ZooKeeper的依赖。
6).参数列表差别很大，例如旧版本的metadata.broker.list和新版本的bootstrp.servers。
KafkaProducer中单参数的send方法仅仅将消息发出而不主管消息发送的结果，及时消息发送失败，也不会得到任何反馈。可能会造成数据的丢失，因此需要使用带回调参数的send方法，即KafkaProducer.send(record,callback)。使用带回调机制的send，消息失败处理逻辑中会显式调用KafkaProducer.close(0),还能防止消息的乱序问题。
Kafka的broker中，如果设置两者相等，只要一个副本出现异常，分区就无法正常工作，虽然提升了持久性，但是极大降低了可用性，因此建议设置replicaiton.factor=min.insyn.replicas+1。
结合上篇，对于kafka的producer端的初步学习就算是告一段落了，了解了producer构造，分区器，拦截器，序列化器等概念，了解了消息压缩，防丢失等策略，了解了多线程并发提高性能设置，接下来会开启对consumser端的学习，进入下一阶段的学习。加油吧。
END
原文作者：DBA修行之路；如有侵权，请联系管理员删除。
