Kafka生产者调优

1. 提高Kafka对生产者吞吐量

从以下参数优化入手:

• batch.size: 批次大小，默认16k
• linger.ms: 等待时间，默认0, 修改为5-100ms
• compression.type: 可选配置值：gzip、snappy、lz4、zstd, 选择snappy
• RecordAccumulator: 缓冲区大小，默认32m, 修改为64m

public static void main(String[] args) {
    Properties properties = new Properties();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.101.105:9092");
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    // 批次大小  默认
    properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16*1024);
    // 等待时间
    properties.put(ProducerConfig.LINGER_MS_CONFIG, 200);
    // 压缩snappy, 
    properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
    // 缓冲区大小
    properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 64*1024*1024);
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
    //2. 发送数据
    Random random = new Random();
    for (int i = 0; i < 100; i++) {
        int i1 = random.nextInt(110);
        kafkaProducer.send(new ProducerRecord<>("hadoop", "personTable", ""+i1));
    }
    //3. 关闭资源
    kafkaProducer.close();
}

2. 提高生产者发送数据可靠性

生产者发送数据流程中, Kafka使用ACK应答机制来确认数据是否落盘，总共有三个级别：

• 0：生产者发送过来的数据，不需要等数据落盘应答
• 1：生产者发送过来的数据，Leader收到数据后应答。
• -1(all)：生产者发送过来的数据，Leader和ISR队列里面的所有节点收齐数据后应答。-1和all等价。

2.1 数据可靠性分析

1. ACK级别为0： 容易丢数据。没有应答机制，容易丢数据👎。
2. ACK级别为1： 可能性丢数据，比如Leader收到Hello数据并应答完成后，还没开始同步副本，Leader挂了，此时新的Leader不会收到Hello的信息，因为生产者已经认为发送成功了。丢失Hello数据。
3. ACK级别为-1： 基本不会丢失数据，但是出现新的问题：Leader收到数据，所有Follower都开始同步数据，但有一个Follower，因为某种故障，迟迟不能与Leader进行同步，导致ACK迟迟等待不能应答。

针对上述问题，Kafka的解决办法是Leader维护了一个动态的in-sync replica set(ISR)，它是Follower+Leader集合(举例leader: 0，isr: 0, 1, 2)。如果Follower长时间未向Leader发送通信请求或同步数据，则该Follower将被踢出ISR。该时间阈值由replica.lag.time.max.ms参数设定，默认30s。例如2超时，(leader: 0, isr: 0,1)。这样就不用等长期联系不上或者已经故障的节点👍。
但是仍然有场景有丢失数据风险：如果分区副本设置为1个，或者ISR里应答的最小副本数量(min.insync.replicas默认为1)设置为1，和ack=1的效果是一样的，仍然有丢数的风险(leader：0，isr:0)。 数据完全可靠条件🏆 = ACK级别设置为-1 + 分区副本≥2 + ISR里应答的最小副本数量≥2

数据可靠性总结

• acks=0，生产者发送过来数据就不管了，可靠性差，效率高；
• acks=1，生产者发送过来数据Leader应答，可靠性中等，效率中等；
• acks=-1，生产者发送过来数据Leader和ISR队列里面所有Follwer应答，可靠性高，效率低；
• 在生产环境中，acks=0很少使用；acks=1，一般用于传输普通日志，允许丢个别数据；acks=-1，一般用于传输和钱相关的数据，对可靠性要求比较高的场景。

编码调优参数加上：

// 设置 acks
 properties.put(ProducerConfig.ACKS_CONFIG, "all");
 // 重试次数 retries，默认是 int 最大值，2147483647
 properties.put(ProducerConfig.RETRIES_CONFIG, 3);

2.2 数据重复分析

在acks=-1的情形下，生产者发送过来的数据，Leader和ISR队列里面的所有节点收齐数据后准备应答。但是如下图Leader挂了，新的Leader会重新接收之前的数据，接收了两份数据，导致数据重复。

3. 生产者推送数据去重

为了避免丢失数据的同时可能会导致数据重复。

3.1 数据传递语义

• 至少一次(At Least Once) = ACK级别设置为-1 + 分区副本≥2 + ISR里应答的最小副本数量≥2
• 最多一次(At Most Once) = ACK级别设置为0

总结：
At Least Once可以保证数据不丢失，但是不能保证数据不重复；
At Most Once可以保证数据不重复，但是不能保证数据不丢失。
精确一次(Exactly Once): 对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失。
Kafka 0.11版本以后，引入了一项重大特性: 幂等性和事务。

3.2 幂等性

1. 幂等性原理
   幂等性就是指Producer不论向Broker发送多少次重复数据，Broker端都只会持久化一条，保证了不重复。
   精确一次(Exactly Once) = 幂等性 + 至少一次(ack=-1 + 分区副本数≥2 + ISR最小副本数量≥2)。
   重复数据的判断标准: 具有<PID, Partition, SeqNumber>相同主键的消息提交时，Broker只会持久化一条。其中PID是Kafka每次重启都会分配一个新的；Partition--表示分区号; Sequence Number是单调自增的。所以幂等性只能保证的是在单分区单会话内不重复。

2. 使用幂等性
   参数enable.idempotence=true即为开启，默认为true。

3.3 生产者事务

危险

开启事务，必须开启幂等性。

3.4 Kafka事务API

// 1 初始化事务
void initTransactions();
// 2 开启事务
void beginTransaction() throws ProducerFencedException;
// 3 在事务内提交已经消费的偏移量(主要用于消费者)
void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) throws ProducerFencedException;
// 4 提交事务
void commitTransaction() throws ProducerFencedException;
// 5 放弃事务(类似于回滚事务的操作)
void abortTransaction() throws ProducerFencedException;

举例代码实操：

public static void main(String[] args) {
    Properties properties = new Properties();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.101.105:9092");
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    // 批次大小
    properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16*1024);
    // 等待时间
    properties.put(ProducerConfig.LINGER_MS_CONFIG, 200);
    // 压缩snappy
    properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
    // 缓冲区大小
    properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 64*1024*1024);
    // ack级别
    properties.put(ProducerConfig.ACKS_CONFIG, "all");
    // 重试次数, 默认Int类型最大值
    properties.put(ProducerConfig.RETRIES_CONFIG, 3);

    // 指定全局事务ID,  只要字符串唯一即可
    properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction_id_001");

    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

    //初始化事务
    kafkaProducer.initTransactions();
    // 开启事务
    kafkaProducer.beginTransaction();
    //2. 发送数据
    Random random = new Random();
    try{
        for (int i = 0; i < 100; i++) {
            int i1 = random.nextInt(110);
            kafkaProducer.send(new ProducerRecord<>("hadoop", "personTable", ""+i1));
        }
        // 提交事务
        kafkaProducer.commitTransaction();
    }catch (Exception e){
        // 回滚事务
        kafkaProducer.abortTransaction();
    }
    //3. 关闭资源
    kafkaProducer.close();
}

4. 数据有序

生产者发送数据到Broker里面，消费者希望消费的数据是有序的，如果一个Topic数据有多个分区，不能保证所有的消息数据有序，只能保证一个分区内数据有序。使用Spark、Flink的窗口可以解决，它是将所有分区消息数据拉取过来重新排序，需要等待消息的全部发送完毕。如果简单使用Kafka, 将消息做成一个分区就可以实现数据有序。

5. 数据乱序

1. kafka在1.x版本之前保证数据单分区有序，条件如下：
   max.in.flight.requests.per.connection=1(不需要考虑是否开启幂等性)。
2. kafka在1.x及以后版本保证数据单分区有序，条件如下：
   • 未开启幂等性
     max.in.flight.requests.per.connection需要设置为1。
   • 开启幂等性
     max.in.flight.requests.per.connection需要设置小于等于5。

原因说明：因为在kafka1.x以后，启用幂等后，kafka服务端会缓存producer发来的最近5个request的元数据，故无论如何，都可以保证最近5个request的数据都是有序的。 图中Request请求通过selector发送给Broker的过程，Request3请求发送失败，势必导致请求接收到的顺序发生改变，Kafka2.x之后版本会将先收到的Request4、Request5先内存中缓存起来，Request1、Request2根据SeqNumber标识允许先落盘，在等待收到Request3后再按照SeqNumber进行Request4、Request5、Request3重排序进行落盘。