大家好，我是考100分的小小码 ，祝大家学习进步，加薪顺利呀。今天说一说kafka修改偏移量offset_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() 方法让消费者继续从轮询里获取新数据
模式二
① 遇到可重试错误时，把错误写入一个独立的主题，然后继续
② 用一个独立的消费者组负责从该主题上读取错误消息，并进行重试
5、长时间处理
有时候要进行比较复杂的处理，暂停轮询的时间不能超过几秒钟。要保持轮询，因为只有在轮询过程中，才能往broker发送心跳。可以使用一个线程池来处理数据，可以让轮询不获取新的数据，直到工作县好吃呢个处理完成。消费者一直保持轮询，心跳正常，就不会发生再均衡。
8、仅一次传递
有的程序不仅是需要“至少一次”（at least-once语义）（意味着没有数据丢失），还需要仅一次（exactly-once）语义。实现一次性语义，最常用的办法就是把结果写入到一个支持唯一键的系统里，比如：k-v存储、关系数据库、ES或者其他数据存储。可以使用主题、分区和偏移量来作为主键，这样，可以碰巧读取到同一个相同的消息，直接覆盖写入就可以了。这种称为幂等性写入。
还有一种，就是使用关系型数据库，HDFS中一些被定义过的原子操作也经常用来达到相同的目的。把消息和偏移量放在同一个事务里，这样让它们保持同步。消费者启动，获取最近处理过的偏移量，调用seek()方法从偏移量位置继续读取数据
参考文件：
「Kafka权威指南」
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
转载请注明出处: https://daima100.com/10074.html
