早在 0.8.2.2 版本，kafka已支持存入消费的 offset 到Topic中，只是那时候默认是将消费的 offset 存放在 Zookeeper 集群中。现在0.10.1.1及以后的版本中，官方默认将消费的offset存储在 Kafka 的Topic中，同时，也保留了存储在 Zookeeper 的接口，通过 offsets.storage 属性来进行设置。

之前版本，Kafka其实存在一个比较大的隐患，就是利用 Zookeeper 来存储记录每个消费者/组的消费进度。虽然，在使用过程当中，JVM帮助我们完成了一些优化，但是消费者需要频繁的去与 Zookeeper 进行交互，而利用ZKClient的API操作Zookeeper频繁的Write其本身就是一个比较低效的Action，对于后期水平扩展也是一个比较头疼的问题。如果期间 Zookeeper 集群发生变化，那 Kafka 集群的吞吐量也跟着受影响。

在新版 Kafka 以及之后的版本，Kafka 消费的offset都会默认存放在 Kafka 集群中的一个叫 __consumer_offsets 的topic中。

当然，其实她实现的原理也让我们很熟悉，利用 Kafka 自身的 Topic，以消费的Group，Topic，以及Partition做为组合 Key。所有的消费offset都提交写入到上述的Topic中。因为这部分消息是非常重要，以至于是不能容忍丢数据的，所以消息的 acking 级别设置为了 -1，生产者等到所有的 ISR 都收到消息后才会得到 ack（数据安全性极好，当然，其速度会有所影响）。所以 Kafka 又在内存中维护了一个关于 Group，Topic 和 Partition 的三元组来维护的 offset 信息，消费者获取的offset的时候会直接从内存中获取。

kafka 提供三种语义的传递：

               1至少一次

               2至多一次

               3较精确一次

首先在 producer 端保证1和2的语义是非常简单的，至少一次只需要同步确认即可（确认方式分为只需要 leader 确认以及所有副本都确认，第二种更加具有容错性），至多一次最简单只需要异步不断的发送即可，效率也比较高。目前在 producer 端还不能保证较精确一次，在未来有可能实现，实现方式如下：在同步确认的基础上为每一条消息加一个主键，如果发现主键曾经接受过，则丢弃。

在 consumer 端，大家都知道可以控制 offset，所以可以控制消费，其实 offset 只有在重启的时候才会用到。在机器正常运行时我们用的是 position，我们实时消费的位置也是 position 而不是 offset。我们可以得到每一条消息的 position。如果我们在处理消息之前就将当前消息的 position 保存到 zk 上即 offset，这就是只多一次消费，因为我们可能保存成功后，消息还没有消费机器就挂了，当机器再打开时此消息就丢失了；或者我们可以先消费消息然后保存 position 到 zk 上即 offset，此时我们就是至少一次，因为我们可能在消费完消息后offset 没有保存成功。而较精确一次的做法就是让 position的保存和消息的消费成为原子性操作，比如将消息和 position 同时保存到 hdfs 上 ，此时保存的 position 就称为 offset，当机器重启后，从 hdfs重新读入offset，这就是较精确一次。

Kafka中的每个partition都由一系列有序的、不可变的消息组成，这些消息被连续的追加到partition中。partition中的每个消息都有一个连续的序号，用于partition唯一标识一条消息。

Offset记录着下一条将要发送给Consumer的消息的序号。

Offset从语义上来看拥有两种：Current Offset和Committed Offset。

Current Offset

Current Offset保存在Consumer客户端中，它表示Consumer希望收到的下一条消息的序号。它仅仅在poll()方法中使用。例如，Consumer第一次调用poll()方法后收到了20条消息，那么Current Offset就被设置为20。这样Consumer下一次调用poll()方法时，Kafka就知道应该从序号为21的消息开始读取。这样就能够保证每次Consumer poll消息时，都能够收到不重复的消息。

Committed Offset

Committed Offset保存在Broker上，它表示Consumer已经确认消费过的消息的序号。主要通过commitSync和commitAsync
API来操作。举个例子，Consumer通过poll() 方法收到20条消息后，此时Current Offset就是20，经过一系列的逻辑处理后，并没有调用consumer.commitAsync()或consumer.commitSync()来提交Committed Offset，那么此时Committed Offset依旧是0。

Committed Offset主要用于Consumer Rebalance。在Consumer Rebalance的过程中，一个partition被分配给了一个Consumer，那么这个Consumer该从什么位置开始消费消息呢？答案就是Committed Offset。另外，如果一个Consumer消费了5条消息（poll并且成功commitSync）之后宕机了，重新启动之后它仍然能够从第6条消息开始消费，因为Committed Offset已经被Kafka记录为5。

总结一下，Current Offset是针对Consumer的poll过程的，它可以保证每次poll都返回不重复的消息；而Committed Offset是用于Consumer Rebalance过程的，它能够保证新的Consumer能够从正确的位置开始消费一个partition，从而避免重复消费。

在Kafka 0.9前，Committed Offset信息保存在zookeeper的[consumers/{group}/offsets/{topic}/{partition}]目录中（zookeeper其实并不适合进行大批量的读写操作，尤其是写操作）。而在0.9之后，所有的offset信息都保存在了Broker上的一个名为__consumer_offsets的topic中。

Kafka集群中offset的管理都是由Group Coordinator中的Offset Manager完成的。

OffsetManager主要提供对offset的保存和读取，每个broker都有一个OffsetManager实例，kafka管理topic的偏移量有2种方式：

1、ZookeeperOffsetManager：调用zookeeper来存储和接收offset

Consumer offset:
/consumers/[groupId]/offsets/[topic]/[partitionId] /offset number

2、DefaultOffsetManager：提供消费者offsets内置的offset管理，即把偏移量提交至kafka内部，主要由offsets.storage参数决定，默认为zookeeper。也就是说如果配置offsets.storage= kafka，则kafka会把这种offsetcommit请求转变为一种Producer，保存至topic为“__consumer_offsets”的log里面。

OffsetManager主要提供对offset的保存和读取，每个broker都有一个OffsetManager实例，kafka管理topic的偏移量有2种方式：

1、ZookeeperOffsetManager：调用zookeeper来存储和接收offset

Consumer offset:

/consumers/[groupId]/offsets/[topic]/[partitionId] /offset number

2、DefaultOffsetManager：提供消费者offsets内置的offset管理，即把偏移量提交至kafka内部，主要由offsets.storage参数决定，默认为zookeeper。也就是说如果配置offsets.storage= kafka，则kafka会把这种offsetcommit请求转变为一种Producer，保存至topic为“__consumer_offsets”的log里面。

offsets是怎么保存到DefaultOffsetManager?

除了将offset作为logs保存到磁盘上,DefaultOffsetManager维护了一张能快速服务于offset抓取请求的consumer offsets表.

broker在offsets表(缓存)中包含仅仅是”offsets topic”的partitions中属于leader partition对应的条目(存储的是offset).

(即这些partitions的replicas只会在他们自己的logs中有offset信息,并不会对应offset表中的条目).

offsets topic的logs的segment大小相比正常的topics设置的很低,默认”offsets.topic.segment.bytes”=10MB.

这样LogCleaner会更快地选出未清理的segments,结果就是能够更频繁地清理. 同样log的大小设置的越小也有助于

减少读取logs的时间,在把读取出来的logs信息放到offsets缓存中也更快(比如重启kafka时会重新加载日志文件).