Kafka丢失数据分析与解决方案

桃扇骨 2022-12-26 10:19 156阅读 0赞

## 1. 在Kafka Broker丢失数据 ##

Broker会将数据写入系统缓存，然后返回确认信息给Producer，如果是单点的Kafka，数据丢失无法避免，原因是只能调节数据刷到硬盘的时间间隔和缓存大小，到调整时间间隔越小、缓存（PageCache）越小时性能会严重下降

结合Producer和多副本可以基本避免数据丢失：

1.  Producer发送请求，主分片Broker收到数据，写入到缓存，然后刷到磁盘上，会同步数据到从分片
2.  从分片收到请求，将数据写入到系统缓存，然后恢复ack
3.  主分片收到从分片的Ack，然后回复Ack给Producer

这个过程中，如果主分片在第一个步骤断电，数据在内存中，Producer收不到ACK会重试，知道达到重试阈值。

这个过程需要权衡配置acks的要求，如果为0，则表示Producer直接写入不等待ACK，数据很可能丢失；如果为1，则表示Producer写入数据到系统Cache，数据在Producer可能丢失，如果为-1，表示要等待所有从分片确认信息，数据则不回丢失

性能考虑和可靠性考虑需要权衡

> 对于可靠性要求很高的场景，应该要求最小副本数（min.insync.replicas），如果没有达到最小副本数，可能存在ack=1的场景，那么数据就可能丢失。

## 2. Producer数据丢失 ##

Producer为了提高发送数据的效率，可以将数据缓存在Buffer中，一次性发送大量数据，那么如果Buffer中数据没有及时处理Producer就因为某些原因退出（OOM、被Kill等）就可能导致数据丢失。或者生产数据过快，导致Buffer装满了还没有发送，也可能导致丢失。

避免或者缓解方式：

1.  Producer同步数据发送，或者采用阻塞的带一定容量上限的线程池
2.  扩大Buffer大小，防止Buffer装满，但是如果程序被异常退出，则数据会丢失
3.  生产的数据在Producer本地落盘，然后再使用另一个程序来发送到Kafka,例如Filebeat、Logstash等

## 3. Consumer端数据丢失 ##

Consumer会从Kafka中接收并处理消息，然后记录offset，提交offset有两种方式，分为手动提交和自动提交，如果是自动提交，就可能存在数据丢失的情况，就是在数据还没有消费成功就已经提交了offset，如果这时Consumer被退出，数据就丢失了。

避免方法：

*  采用手动提交方式可以避免出现上述问题，但会造成数据被重复消费

自动提交：

package com.zjw.consumer;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    
    import java.util.Arrays;
    import java.util.Properties;
    
    /** * Created by zjwblog<co.zjwblog@gmail.com> on 2020/12/4 */
    public class Consumer1 { 
      public static void main(String[] args) { 
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test");
        // 开启自动提交
        props.put("enable.auto.commit", "true");
        // 自动提交的时间间隔，此处是1000ms
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("foo", "bar"));
        while (true) { 
          // 调用poll后，1000ms后，消息状态会被改为 committed
          ConsumerRecords<String, String> records = consumer.poll(100);
          for (ConsumerRecord<String, String> record : records) { 
            insertIntoDB(record); // 消息将被处理，时间可能会超过1000ms
          }
        }
      }
    
      private static void insertIntoDB(ConsumerRecord<String, String> record) { 
        // 数据处理逻辑
      }
    }

收到方式提交：

package com.zjw.consumer;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    
    import java.util.ArrayList;
    import java.util.Arrays;
    import java.util.List;
    import java.util.Properties;
    
    /** * Created by zjwblog<co.zjwblog@gmail.com> on 2020/12/4 */
    public class Consumer2 { 
      public static void main(String[] args) { 
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test");
        // 关闭自动提交
        props.put("enable.auto.commit", "false");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("foo", "bar"));
        final int minBatchSize = 200;
        List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
        while (true) { 
          ConsumerRecords<String, String> records = consumer.poll(100);
          for (ConsumerRecord<String, String> record : records) { 
            buffer.add(record);
          }
          if (buffer.size() >= minBatchSize) { 
            insertIntoDb(buffer);
            // 所有消息消费完毕以后，才进行commit操作
            consumer.commitSync();
            buffer.clear();
          }
        }
      }
    
      private static void insertIntoDb(List<ConsumerRecord<String, String>> buffer) { 
        // 数据处理逻辑
      }
    }