Kafka实战之生产者实践

本文分享一下我了解关于Kafka的一些最佳实践。

1. 合理配置参数减少KafkaProducer初始化时候的TCP连接
   Kafka中所有的通信都是基于TCP的，KafkaProducer实例在被创建之后，在后台会开启一个sender线程，这个线程会建立与broker之间的TCP连接，这些broker是bootstrap.servers指定的所有broker，这个参数指定了producer启动时需要建立链接的broker，这里强调的是producer启动时，设想一下如果该参数设置了成百上千的broker，那producer初始化时需要和成百上千的broker建立TCP链接，这无疑是一个非常耗时的操作，通常建议该参数指定3～4个broker即可，因为只要连上集群中的任意一台broker都能获取到整个集群的信息。

2. 适时开启消息压缩
   Kafka提供了消息压缩的功能，生产端开启消息压缩只需要设置compression.type为指定的压缩方式就可以了，采用压缩其实就是用时间换空间的思路，考虑是否开启消息压缩，应该考虑两个方面的因素，一是生产者CPU的使用情况，如果CPU的负载比较高，那么开启消息压缩对于CPU来说无疑是雪上加霜；二是带宽，如果带宽资源比较吃紧，此时可以考虑开启消息压缩来减小消息的大小，从而提高网络传输的效率。是否开启消息压缩需要综合考虑CPU资源和带宽资源，如果CPU资源比较充足，而带宽吃紧，应该开启消息压缩，反之则应该关闭压缩。

1. 消息丢失场景分析
   Kafka保证对已提交的消息做有限度的持久化，首先针对的是已提交的消息，当一条消息被一定数量的broker成功接收并写入日志文件了，这些broker会告诉producer这条消息已成功提交，此时消息的状态才是已提交，如果发生网络抖动消息并没有到达broker或者消息太大超过了参数设置的最大消息大小被broker拒绝了，那么消息并未成功提交，为成功提交的消息，Kafka不作持久化保证，那么如果选取producer.send(message)这种不带回调函数的API，Kafka消息发送是异步的，send之后直接返回，那么producer这边不知道消息是否成功发送，也没有做补偿措施，就会导致消息“丢失”。

2. 参数配置
   为了保证消息不丢失，还需要配置一些Kafka相关的参数

• acks： acks是生产端的一个参数，控制对消息“已提交”的定义，可选的配置有0、1、all(或-1)，0表示认为消息发送出去默认成功的直接进行提交，1表示当Leader副本成功接收消息并写入log，该消息就可以认为是已提交的，all表示消息必须所有ISR列表中和Leader同步的副本都成功接收消息并成功写入log中，消息才能算是已提交，故要严格保证消息无丢失，请将acks的参数设置为all，但是此时吞吐量会受影响；
• retries: 设置 retries 为一个较大的值，retries表示当出现网络抖动，消息无法发送时，会进行重试，避免网络抖动带来的影响，可以将该值设置比较大；
• unclean.leader.election.enable: 该参数控制哪些副本能够参与竞选分区的Leader，如果一个Broker落后原来的Leader太多，如果它成为新的Leader，就会造成那些落后的消息的丢失，所以应该将这个参数设置为false，既不允许落后于Leader的副本参与Leader的选举；
• replication.factor: 这个参数表示的每个分区有多少个副本，副本的作用是做数据冗余来进行容灾，至少设置该值大于等于3，将消息多保存几份；
• min.insync.replicas: 这个参数控制一条消息需要写入多少个副本才算已提交，默认值为1，在生产环境应该设置为大于的数来提升消息的持久性；
  注意：replication.factor的值应该要大于min.insync.replicas的值，如果两个值相等，如果一个broker宕机后会导致整个分区不可用，此时应该根据业务需要在高可用和持久性中权衡。

3. API选择
   选择带有回调的send(message, callback)发送消息，通过回调函数可以知道消息发送成功还是发送失败，知道发送失败后，进行补偿措施来避免损失。

既然是消息系统，我们必须考量消息的可靠性，即每条消息是否成功发送，消息是否重复等，通常对消息可靠性的保证包括：

• 最多一次
• 至少一次
• 精确一次

Kafka默认情况下确保消息至少发送一次，会出现消息重复发送的情况，如消息发送给broker后，broker将消息写入了log，但是在发送给producer确认消息时出现了网络抖动，producer没有收到broker的提交响应，认为这条消息并没有成功写入broker，会自动尝试重新发送，就会造成消息重复问题。要实现消息发送精确一次的语义，必须借助Kafka提供的幂等和事务生产者。

1. 幂等生产者
   幂等起初是一个数学概念，表示某某些操作或者函数能够被执行多次，并且每次执行的结果都是一样的，比如绝对值函数abs和取整函数floor等，后来延伸到计算机领域，表示某个系统的子系统的操作不会影响系统整体的状态，幂等性带来的好处是可以安全的重复任何幂等操作。
   Kafka通过设置“enable.idempotence”为true来开启幂等生产者，即通过

props.put("enable.idempotence",true);
//或者
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true)

Kafka的幂等生产者只能保证单分区和单会话的消息幂等性，这与Kafka内部幂等消息的实现有关，不作展开。

2. 事务生产者

   
   
   

   前面提到幂等生产者只能保证单分区和单会话的消息幂等性，如上图所示，producer发送消息给broker0上面的partition0，消息成功写入log文件后，broker0宕机，无法发送确认提交，producer没有收到确认消息，会重试，而此时负载均衡将该消息发送给了broker1上面的partition1，那么该消息也出现了重复发送的问题。要实现“精确一次”则必须使用采用更加严格的事务型Producer。
   事务Producer的开启需要在生产端设置transactional.id参数，可以为其设置一个与业务相关具有具体意义的名字，同时也需要在代码中开启、提交和回滚事务

transactionalProducer.initTranscations();
try {
    transactionalProducer.beginTransaction();
    transactionalProducer.send(message1);
    transactionalProducer.send(message2);
    transactionalProducer.commitTransaction();
} catch (KafkaException e) {
    transactionalProducer.abortTranscation()
}

message1和message2作为一个事务统一提交给Kafka，两条消息要么全部提交成功要么全部提交失败，同时还需要设置消费端的事务隔离级别为read_committed。