在 [Kafka 101 - 5] 图文并茂地介绍 offset 概念 中我们介绍了消息的 offset 和消费者的 offset，并且提到消费者的 offset 的维护方式是消费者自己提交到一个特殊 topic，听起来似乎很简单，但实际这个提交 offset 的过程也有点内容，所以本文来学习 Kafka 中消费者提交 offset 的五种方式。

需要明确的一点是，即使 Kafka 中维护了消费者的 offset，消费者仍然有可能重复消费或者少消费数据的，如果想要保证消费数据的完全准确，不丢不重，即所谓的“Exactly Once”，需要使用别的机制来保证，比如 Flink 的 checkpoint 机制，但这些机制也需要能够正确的理解 offset 的提交。

内容提要：

1. 环境说明
2. 自动提交
3. 同步提交
4. 异步提交
5. 同步&异步结合
6. 提交指定 offset
7. 总结

1. 环境说明

操作系统：MacOS/Linux
Kafka：本地安装了社区版的 2.3.1 版本，运行在 9092 端口

Kafka 的安装使用可以参考：[Kafka 101-1] Kafka安装使用
消费者 Java API 的基本使用可以参考：[Kafka 101-3] 使用Java API消费数据实战

2. 自动提交

这是 offset 提交的默认方式。

消费者有一个参数叫做 enable.auto.commit ，表示是否启用 offset 的自动提交，默认值为 true，并且还有一个配套的参数叫做 auto.commit.interval.ms，表示自动提交 offset 的时间间隔，默认值是 5000，即 5s 提交自动提交一次 offset，还记得消费者的 poll 循环吗？长这个样子：

while (true) {
  // consumer 是一个 KafkaConsumer对象
  ConsumerRecords<String, String> records = consumer.poll(100);
  for (ConsumerRecord<String, String> record : records) {
    // 用 println 模拟数据数据过程
    System.out.printf("消息内容为：%s\n", record.value());
  }
}

每当调用 KafkaConsumer 对象的 poll 方法时，它会去检查距离上一次提交 offset 是否已经过去 5s，如果够 5s 的话，则会帮助你把这次 poll 操作消费的最大 offset 提交给 Kafka broker。

注意：如果在自动提交了 offset 之后 3s 的时候程序因为某种原因挂掉了，并且在这 3s 期间消费了 10000 条数据，那么当程序重启后，这 10000 条数据会被再次消费，因为 Kafka 中记录的是最后一次提交的 offset，程序重启后会从这个 offset 开始继续消费，所以，自动提交有重复消费数据的风险，而且我们完全无法控制，看起来不是很棒，所以在重视数据准确性的场景中，都不会采用自动提交的方式。

下文中的方法都需要在构造 KafkaConsumer 对象的时候传入 auto.commit.offset 参数，并且设置为 false。

3. 同步提交

直接上代码：

while (true) {
  ConsumerRecords<String, String> records = consumer.poll(100);
  for (ConsumerRecord<String, String> record : records) {
    // 用 println 模拟数据处理过程
    System.out.printf("消息内容为：%s\n", record.value());
  }
  // 处理完毕后提交 offset
  try {
    consumer.commitSync();
  } catch (CommitFailedException e) {
    // commitSync 会一直尝试提交直到成功或者遇到不可恢复的错误
    log.error("commit faile", e)
  }
}

可以看到，同步提交 offset 的方法叫做 commitSync()，这个方法会提交最近一次调用 poll 所消费到的最大 offset，这上面的例子中，我们是把 commitSync() 放在了数据处理之后，如果程序在数据处理过程中挂掉，这时已经处理了一部分数据（比如写到了MySQL 中），那么重启后会从上一次提交的 offset 继续消费，之前已经写入 MySQL 的数据会被再处理一次，因此MySQL 中的数据会重复。

那么如果我们把 commitSync 放在数据处理的代码之前呢，答案是数据有可能丢失，因为这种情况下的问题是，可能已经提交了 offset，但是数据处理过程没有完成，故障重启之后只能消费到新数据，重启前没有来得及处理的数据也处理不到了。

所以，同步提交的方式，有可能造成数据重复，也有可能造成数据丢失，取决于 commitSync() 方法的位置。

4. 异步提交

有个同步，就有个异步，同步相比异步的不足是，提交 offset 的时候程序会阻塞住，限制了消费吞吐量，因此就有了异步提交，上代码：

while (true) {
  ConsumerRecords<String, String> records = consumer.poll(100);
  for (ConsumerRecord<String, String> record : records) {
    // 用 println 模拟数据处理过程
    System.out.printf("消息内容为：%s\n", record.value());
  }
  // 处理完毕后提交 offset
  consumer.commitAsync();
}

调用了 commitAsync 方法后，程序会继续执行，确实可以提高吞吐量，但是没有只有好处没有坏处的事，上面提到，提交 offset 是有可能失败的，同步提交方法会一直尝试提交，要么成功，要么遇到不可恢复的错误抛异常，但是异步提交方法不会重试，为什么不重试呢？因为如果不成功就一直重试，有可能更新一次的 offset 提交都已经完成了，如果重试成功了，反而会把更新的 offset 给覆盖掉，造成重复，所以干脆不重试。

异步的方法一般都一个回调函数，这个也有：

consumer.commitAsyn(new OffsetCommitCallback() {
    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) {
        if (e != null)
            log.error("Commit failed for offsets {}", offsets, e);
    }
})

如果想要在异步提交失败的时候重试，《Kafka 权威指南》给了一种方法，维护一个单调递增的序列号，每提交一次，递增一下这个序列号，并把序列号传给回调函数，当在回调函数中发现失败时，比较一下回调函数中的序列号和当前序列号的大小，如果当前的序列号已经比回调函数的序列号大，那就不用重新提交 offset 了，因为已经有一个更新的 offset 在提交了。

我觉得按照这个说法，代码可能长这个样子（声明：没有验证过）：

// 维护全局序列号
int global_seq = 0;
while (true) {
  ConsumerRecords<String, String> records = consumer.poll(100);
  for (ConsumerRecord<String, String> record : records) {
    System.out.printf("消息内容为：%s\n", record.value());
  }
  // 异步提交 offset 时传给回调函数
  consumer.commitAsync(new MyOffsetCommitCallback(global_seq++));
}
// 私有类实现 OffsetCommitCallback 接口
private class MyOffsetCommitCallback implements OffsetCommitCallback {
  private int seq;
  // 回调函数初始化时记录当前序列号
  MyOffsetCommitCallback(int global_seq) {
    this.seq = global_seq;
  }
  public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) {
        if (exception != nul) {
      // 两者相等表示还没有更新的 offset 在提交，可以重试
      if (this.seq == global_seq) {
        // 在这里重试，怎么重试你来想吧
      }
    }
  }
}

5. 同步&异步结合

其实偶尔有一两次提交 offset 失败问题不大，因为只要后续有 offset 提交成功了，之前的失败可以忽略的。但是如果知道这是最后一次提交了，那么还是有必要确保这次提交能成功的。所以一种常用的提交 offset 的方式，是同时使用同步提交和异步提交，它可以兼顾效率和可靠性（数据准确性），上代码：

try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
      System.out.printf("消息内容为：%s\n", record.value());
        }
    // 即使失败也不要紧，要么有下一次异步的提交，要么有关闭前的最后一次提交
    consumer.commitAsync();
  }
// 捕获处理不了的异常
} catch (Exception e) {
  log.error("Unexpected error", e);
} finally {
  try {
    // 关闭 consumer 前最后一次提交使用同步的方式，最大程度的确保成功
    consumer.commitSync();
  } finally {
    consumer.close();
  }
}

简单的解释一下代码：poll 循环里用异步提交，效率高，整个 poll 循环捕获到异常之后在关闭前进行一次同步提交，稳妥，保证最新的 offset能被提交（当然，如果是不可恢复的异常，比如 Kafka 宕机，发生这种异常是无法提交成功的）。

6. 提交指定 offset

上面的这些提交方式，都是 poll 一次，提交一次，其实还可以 poll 一次，提交多次，比如每处理一条数据，提交一次，上代码：

// 引入由 TopPartition 和 OffsetAndMetadata 组成的 Map 类型
private Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
while (true) {
  ConsumerRecords<String, String> records = consumer.poll(100);
  for (ConsumerRecord<String, String> record : records)
  {
    System.out.printf("消息内容为：%s\n", record.value());
    // 更新当前分区的 offset
    currentOffsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset()+1, "no metadata"));
    // 这里用了异步提交，也可以用同步提交
    // consumer.commitSync(currentOffsets);
    consumer.commitAsync(currentOffsets, null);
  }
}

解释一下：同步提交的方法和异步提交的方法都可以接受Map<TopicPartition, OffsetAndMetadata> 类型的参数，提交这个参数指定的 offset，而不是通过 poll 方法消费的最大 offset。

7. 总结

各种 offset 的提交方式和优缺点总结如下：

提交方式	优点&缺点
自动提交	可能有大量重复消费，不受控制
同步提交	效率低，也有可能重复消费， 但比自动提交少
异步提交	调用效率高
同步&异步结合	高吞吐量，且可靠，可能有少量的重复消费（推荐）
提交指定 offset	可以比其它方式更加频繁的提交，仍然有可能重复

欢迎交流讨论，吐槽建议。

勤学似春起之苗，不见其增，日有所长
辍学如磨刀之石，不见其损，日有所亏
关注【大数据学徒】，用技术干货助你日有所长

大数据学徒