Kafka offset management

0、offset管理

  • offset是一个long型数值,唯一对应一条message。
  • 消费者会向__consumer__offset的这个特殊topic提交偏移量。具体group的消费情况,记录在这个topic的哪个partition,计算公式如下:
brokerId = Math.abs(groupId.hashCode()) % partitionCount 
// parititionCount 是__consumer_offsets的 partition 数,默认是50个

1、自动提交offset

设置enable.auto.commit为true,则会自动提交偏移量
。consumer在调用poll()方法的时候,会检查{auto.commit.interval.ms}(默认5s)的间隔,如果满足则会自动提交offset。但是可能导致消息被重复消费。例如每5s提交偏移量一次,但是在第3s的时候发送了再均衡,这个时候偏移量不会被提交上去,但是之前的3s消息就会被重复消费。

2、手动提交

(1)commitSync,同步提交

commitSync()函数提交偏移量并阻塞等待server端响应。如果没有发生不可恢复的错误,commitSync()会一直尝试直到成功

    while(true) {
        ConsumerRecords records = consumer.poll(100);
        
        for (ConsumerRecord record : records) {
        }
        try {
            consumer.commitSync();
        } catch (Exception e) {
        }
    }
(2)commitAsync(),异步提交
  • commitAsync()函数会异步提交请求,但是commitAsync()只会尝试一次,如果失败就不会再发送。因为可能在第一次提交偏移量失败以后,消费者消费了新的一批数据并准备提交offset,如果新的offset提交成功(新的offset比旧的offset大),旧的offset再提交成功,这个时候发生再均衡,就回出现重复消息。commitAsync()支持回调函数。
(3)同步与异步组合提交
      try {
        AtomicInteger atomicInteger = new AtomicInteger(0);
        while (true) {
            ConsumerRecords records = consumer.poll(5);
            for (ConsumerRecord record : records) {
            }
            consumer.commitAsync(new OffsetCommitCallback() {
                private int marker = atomicInteger.incrementAndGet();
                @Override
                public void onComplete(Map offsets,
                                       Exception exception) {
                    // 当发生错误的时候要重试的时候进行校验。利用上面的atomicInteger的incrementAndGet方法进行递增保证,当失败的时候check本次提交是否是最新的提交。
                    if (exception != null) {
                        if (marker == atomicInteger.get()) consumer.commitAsync(this);
                    }
                }
            });
        }
    } catch (WakeupException e) {
        // ignore for shutdown
    } finally {
        consumer.commitSync(); //Block
        consumer.close();
        System.out.println("Closed consumer and we are done");
    }
(4)提交特定offset
        Map currentOffsets = Maps.newHashMap();
        int count = 0;
        try {
            while(true) {
                ConsumerRecords records = consumer.poll(100);
                for (ConsumerRecord record : records) {
                    System.out.printf("topic = %s,partition = %s,offset = %s,customer = %s",
                            record.topic(),record.partition(),record.offset(),record.key());
                    currentOffsets.put(new TopicPartition(record.topic(),record.partition()),
                            new OffsetAndMetadata(record.offset() + 1));
                    if( count % 1000 == 0) {
                        // no callback
                        consumer.commitAsync(currentOffsets,null);
                    }
                    count ++;
                }
            }
        } finally {
            try {
                consumer.commitSync();
            } finally {
                consumer.close();
            }
        }

3、offset缓存

在broker上,有一份关于offset的缓存,记录了(topic,groupid)维度的最新offset信息。consumer client在commit offset的时候,offset会同步更新这份缓存以及写入对应的log文件中。当这台broker宕机的时候,对应的partition-replica在的broker就会从log文件中重建缓存。

你可能感兴趣的