「Kafka」Kafka中offset偏移量提交

在消费kafka中分区的数据时，我们需要跟踪哪些消息是读取过的、哪些是没有读取过的。这是读取消息不丢失的关键所在。

Kafka是通过offset顺序读取事件的。如果一个消费者退出，再重启的时候，它知道从哪儿继续读取消息进行处理。所以，消费者需要「提交」属于它们自己的偏移量。如果消费者已经提交了偏移量，但消息没有得到有效处理，此时就会造成消费者消息丢失。所以，我们应该重视偏移量提交的时间点以及提交的方式。

Kafka消费者的可靠性配置

1、group.id

• 如果两个消费者有相同的 group.id，并且定义同一个主题，那么每个消费者都会消费一个分区的数据

2、auto.offset.reset

• 这个参数的作用是：当没有偏移量提交（例如：消费者第一次启动、或者请求的偏移量在broker上不存在时），消费者会如何处理
• earliest：消费者从分区的开始位置读取大量的重复数据，可以保证个最少的数据丢失
• latest：消费者会从分区的末尾开始读取数据，可以减少重复读，但很有可能会错过一些消息

3、enable.auto.commit

• 可以设置自动提交偏移量，可以在代码中手动提交偏移量
• 自动提交，可以让消费者逻辑更简单
• 但它无法控制重复处理消息、或者如果消息交给另外一个后台线程去处理，自动提交机制可能会在消息还没有处理完就提交了偏移量

4、auto.commit.interval.ms

• 通过该参数，可以配置提交的频率。默认：每5秒钟提交一次
• 提交的频率高，也是会增加额外的开销的

显示提交偏移量

如果我们希望能够更有效地控制偏移量提交的时间点，就需要显示地提交偏移量。

1、总是在处理完事件后再提交偏移量

如果所有的处理都是在轮询里完成，无需在轮询之间维护状态，那么可以使用自动提交，或者在轮询结束后进行手动提交。

2、提交频率是性能和重复消息数量之间的权衡

这个意思是：提交频率越高，重复消息处理的数量越少，性能也是比较低的。提交频率越低，重复消息处理的数量越多，性能是比较好的。所以，要根据实际的情况，来衡量在什么时机，来提交偏移量。即使是在最简单的场景你，也需要在一个循环中多次提交偏移量。

3、确保对提交的偏移量心里有数

一定要在处理完消息后，再提交偏移量，否则会出现某些消息会被处理。

4、消费者可能需要重试

但处理消息出现问题时，例如：把Kafka中的数据写入到HBase中，此时HBase临时不可用。我们想要重试。假设这条消息是：#30，#30处理失败了。那大家想想？#31能提交吗？

显然是不能的，如果#31提交了，那么#31之前的所有数据，都不会被处理了。我们可以使用以下几种模式来处理：

模式一

① 但遇到可重试错误时，提交最后一个处理成功的偏移量

② 把没有处理好的消息保存到缓冲区

③ 调用 pause() 方法，确保其他的轮询不会返回数据

④ 尝试重新处理缓存中的数据，如果重试成功，或者重试次数达到上限并决定放弃，把错误记录下来并丢弃消息

⑤ 调用 resume() 方法让消费者继续从轮询里获取新数据

模式二

① 遇到可重试错误时，把错误写入一个独立的主题，然后继续

② 用一个独立的消费者组负责从该主题上读取错误消息，并进行重试

长时间处理

有时候要进行比较复杂的处理，暂停轮询的时间不能超过几秒钟。要保持轮询，因为只有在轮询过程中，才能往broker发送心跳。可以使用一个线程池来处理数据，可以让轮询不获取新的数据，直到工作县好吃呢个处理完成。消费者一直保持轮询，心跳正常，就不会发生再均衡。

仅一次传递

有的程序不仅是需要“至少一次”（at least-once语义）（意味着没有数据丢失），还需要仅一次（exactly-once）语义。实现一次性语义，最常用的办法就是把结果写入到一个支持唯一键的系统里，比如：k-v存储、关系数据库、ES或者其他数据存储。可以使用主题、分区和偏移量来作为主键，这样，可以碰巧读取到同一个相同的消息，直接覆盖写入就可以了。这种称为幂等性写入。

还有一种，就是使用关系型数据库，hdfs中一些被定义过的原子操作也经常用来达到相同的目的。把消息和偏移量放在同一个事务里，这样让它们保持同步。消费者启动，获取最近处理过的偏移量，调用seek()方法从偏移量位置继续读取数据

参考文件：

「Kafka权威指南」