锐客网

  1. 首页 > it技术 > >

kafka|kafka 提交offset

每次调用poll()方法，它总是返回由生产者写入kafka但还没有被消费者读取过的记录，我们因此可以追溯到哪些记录是被群组里的那个消费者读取的。之前已经讨论过，Kafka不会像其他的JMS队列那样需要得到消费者的确认，这是kafka的一个独特之处。相反，消费者可以使用kafka来追踪消息在分区的位置（偏移量）
那么消费者是如何提交offset的呢？consumer 往一个叫做_consumer_offset的特殊主题发送消息，消息里面包含每个分区的偏移量。如果消费者一直处于运行状态，那么偏移量就没有什么用处。不过，如果消费者发生崩溃或者有新的消费者加入群组，就会触发rebanlance(再均衡)，完成在均衡之后，每个消费者可能分配到新的分区，而不是之前处理的那个，为了能够继续之前的工作，消费者需要读取每个分区最后一次提交的偏移量，然后从偏移量指定的地方继续处理。
Q1 如果提交的偏移量小于客户端处理的最后一个消息的offset，则两者之间的数据就会被重复消费。
Q2 如果提交的偏移量大于客户端处理的最后一个消息的offset,则两者职期间的数据就会丢失。
所以，偏移量的提交对客户端有很大的影响。
自动提交
最简单的方式就是consumer自动提交offset，如果enable.auto.commit =true，那么每过5s,consumer会自动把poll()方法接收到的最大offset提交上去。提交时间间隔由auto.commit.interval.ms 控制，默认是 5s.与消费者里其他的东西一样，自动提交也是在轮询里进行的。consumer每次在进行查询的时候回检查是否该提交偏移量了，如果是，那么就会提交从上一次轮询返回的偏移量。
不过， 在使用这种渐变的方式之前，需要知道它将会带来怎样的后果。
假设我们使用默认的5s提交时间间隔，在最近一次提交之后的3是，发生了在均衡，在均衡之后，消费者从最后一次提交的offset的位置开始读取消息，这个时候offset已经落后了3s,所以在这3s到达的消息会被重复处理。可以通过修改提交时间来频繁的提交offset，减少可能出现重复消息的时间窗，不过这种情况是无法完全避免的。
同步提交
处理完当前批次的消息，在轮询更多的消息之前，调用commitSync方法提交当前批次最新的offset
只要没有发生不可恢复的错误，commitSync()会一直尝试直至提交成功，如果提交失败，我们也只能把异常记录到日志里。
异步提交
【kafka|kafka 提交offset】提交一个offset，然后继续做其他事情，如果提交失败，错误信息和偏移量会被记录下来。commitAsync和commitSync不同在于，它不会一直重试，是因为有可能在它收到服务器响应之前，可能有一个更大的offset已经提交成功。另外commitAsync支持回调。
本文摘自kafka权威指南

    推荐阅读


    上一篇：一树黎明一树春

    下一篇：“我只管你”