Kafka 位移

Consumer位移管理机制

将Consumer的位移数据作为一条条普通的Kafka消息，提交到__consumer_offsets中。可以这么说，__consumer_offsets的主要作用是保存Kafka消费者的位移信息。使用Kafka主题来保存位移。

消息格式

位移主题就是普通的Kafka主题。也是一个内部主题，但它的消息格式却是Kafka自己定义的KV对（Key和Value分别表示消息的键值和消息体），用户不能修改，Kafka Consumer有API去提交位移，也就是向位移主题写消息。不要自己写个Producer随意向该主题发送消息。

主题消息的Key中应该保存标识Consumer的字段，也就是Consumer Group的Group ID，标识唯一的Consumer Group，因为Consumer提交位移是在分区层面上进行的，即它提交的是某个或某些分区的位移，那么很显然，Key中还应该保存 Consumer要提交位移的分区。

总结：位移主题的Key中应该保存3部分内容：<Group ID，主题名，分区号>。

还有2种格式：

1. 用于保存Consumer Group信息的消息，用来注册Consumer Group

2. tombstone消息，即墓碑消息，也称delete mark：用于删除Group过期位移甚至是删除Group的消息。

位移主题的创建

当Kafka集群中的第一个Consumer程序启动时，Kafka会自动创建位移主题。

分区数是怎么设置的呢？这就要看Broker端参数offsets.topic.num.partitions的取值了。它的默认值是50，因此Kafka会自动创建一个50分区的位移主题。Broker端另一个参数offsets.topic.replication.factor 控制副本数，默认为3。所以：如果位移主题是Kafka自动创建的，那么该主题的分区数是50，副本数是3。

提交位移（Committing Offsets）

Consumer需要向Kafka汇报自己的位移数据，这个汇报过程被称为提交位移（Committing Offsets）。当Consumer发生故障重启之后，就能够从Kafka中读取之前提交的位移值，然后从相应的位移处继续消费，从而避免整个消费过程重来一遍。

从用户的角度来说，位移提交分为自动提交和手动提交；从Consumer端的角度来说，位移提交分为同步提交和异步提交。

Kafka Consumer提交位移的方式有两种：自动提交位移和手动提交位移。

手动提交位移

enable.auto.commit 如果值是false，则为手动提交，它能够把控位移提交的时机和频率。可以使用Kafka Consumer API的consumer.commitSync等方法，当调用这些方法时，Kafka会向位移主题写入相应的消息。

while (true) {
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofSeconds(1));
process(records); // 处理消息
try {
consumer.commitSync();
} catch (CommitFailedException e) {
handle(e); // 处理提交失败异常
}
}

调用consumer.commitSync()方法的时机，是在处理完了poll()方法返回的所有消息之后。如果过早提交了位移，就可能会出现消费数据丢失的情况。它还也有一个缺陷，就是在调用commitSync()时，Consumer程序会处于阻塞状态，直到远端的Broker返回提交结果，这个状态才会结束，影响整个应用程序的TPS。

Kafka社区为手动提交位移提供了另一个API方法：KafkaConsumer#commitAsync() ，这是一个异步操作。调用commitAsync()之后，它会立即返回，不会阻塞，因此不会影响Consumer应用的TPS。由于它是异步的，Kafka提供了回调函数（callback），在实现提交之后的逻辑，比如记录日志或处理异常等。下面这段代码展示了调用commitAsync()的方法：

while (true) {
ConsumerRecords<String, String> records =
consumer.poll(Duration.ofSeconds(1));
process(records); // 处理消息
consumer.commitAsync((offsets, exception) -> {
if (exception != null)
handle(exception);
});
}

commitAsync是否能够替代commitSync呢？

答案是不能。commitAsync的问题在于，出现问题时它不会自动重试。因为它是异步操作，倘若提交失败后自动重试，那么它重试时提交的位移值可能早已经“过期”或不是最新值了。因此，异步提交的重试其实没有意义，所以commitAsync是不会重试的。

将commitSync和commitAsync组合使用

try {
    while(true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
        process(records); // 处理消息
        commitAysnc(); // 使用异步提交规避阻塞
    }
} catch(Exception e) {
    handle(e); // 处理异常
} finally {
    try {
        consumer.commitSync(); // 最后一次提交使用同步阻塞式提交
    } finally {
        consumer.close();
    }
}

对于常规性、阶段性的手动提交，我们调用commitAsync()避免程序阻塞，而在Consumer要关闭前，我们调用commitSync()方法执行同步阻塞式的位移提交，以确保Consumer关闭前能够保存正确的位移数据。将两者结合后，既实现了异步无阻塞式的位移管理，也确保了Consumer位移的正确性。

分批处理（细粒度的位移提交）

commitSync(Map<TopicPartition, OffsetAndMetadata>)

commitAsync(Map<TopicPartition, OffsetAndMetadata>)

它们的参数是一个Map对象，键就是TopicPartition，即消费的分区，而值是一个OffsetAndMetadata对象，保存的主要是位移数据。

例如：如何每处理100条消息就提交一次位移呢？以commitAsync为例，展示一段代码，实际上，commitSync的调用方法和它是一模一样的。

private Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
int count = 0;
// 其他操作
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
    for (ConsumerRecord<String, String> record: records) {
        process(record); // 处理消息
        offsets.put(new TopicPartition(record.topic(), record.partition()) , new OffsetAndMetadata(record.offset() + 1)；
        if（count % 100 == 0）
            consumer.commitAsync(offsets, null); // 回调处理逻辑是null
            count++;
        }
    }
}

程序先是创建了一个Map对象，用于保存Consumer消费处理过程中要提交的分区位移，之后开始逐条处理消息，并构造要提交的位移值。要提交下一条消息的位移，这里构造OffsetAndMetadata对象时，使用当前消息位移加1的原因。代码的最后部分是做位移的提交。这里设置了一个计数器，每累计100条消息就统一提交一次位移。与调用无参的 commitAsync不同，这里调用了带Map对象参数的commitAsync进行细粒度的位移提交。这样，这段代码就能够实现每处理100条消息就提交一次位移，不用再受poll方法返回的消息总数的限制了。

自动提交位移

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "2000");
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) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records)
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}

Consumer端有个参数叫enable.auto.commit，如果值是true，则Consumer 定期提交位移，提交间隔由一个专属的参数auto.commit.interval.ms来控制。但是没法把控Consumer端的位移管理。

一旦设置了enable.auto.commit为true，Kafka会保证在开始调用poll方法时，提交上次poll返回的所有消息。从顺序上来说，poll方法的逻辑是先提交上一批消息的位移，再处理下一批消息，因此它能保证不出现消费丢失的情况。但自动提交位移的一个问题在于，它可能会出现重复消费。

在默认情况下，Consumer每5秒自动提交一次位移。现在，我们假设提交位移之后的3秒发生了Rebalance操作。在Rebalance之后，所有Consumer从上一次提交的位移处继续消费，但该位移已经是3秒前的位移数据了，故在Rebalance发生前3秒消费的所有数据都要重新再消费一次。虽然能够通过减少auto.commit.interval.ms的值来提高提交频率，但这么做只能缩小重复消费的时间窗口，不可能完全消除它。这是自动提交机制的一个缺陷。

自动提交位移问题：

自动提交位移，那么就可能存在一个问题：只要Consumer一直启动着，它就会无限期地向位移主题写入消息。

假设Consumer当前消费到了某个主题的最新一条消息，位移是100，之后该主题没有任何新消息产生，故Consumer无消息可消费了，所以位移永远保持在100。由于是自动提交位移，位移主题中会不停地写入位移=100的消息。显然Kafka只需要保留这类消息中的最新一条就可以了，之前的消息都是可以删除的。这就要求Kafka必须要有针对位移主题消息特点的消息删除策略，否则这种消息会越来越多，最终撑爆整个磁盘。

Kafka使用Compact策略来删除位移主题中的过期消息，避免该主题无限期膨胀。那么应该如何定义Compact策略中的过期呢？对于同一个Key的两条消息M1和M2，如果M1的发送时间早于 M2，那么M1就是过期消息。Compact的过程就是扫描日志的所有消息，剔除那些过期的消息，然后把剩下的消息整理在一起。