Kafka中的每個partition都由一系列有序的猜嘱、不可變的消息組成毅整，這些消息被連續(xù)的追加到partition中映胁。partition中的每個消息都有一個連續(xù)的序號预侯，用于partition唯一標(biāo)識一條消息致开。
Offset記錄著下一條將要發(fā)送給Consumer的消息的序號。
Offset從語義上來看擁有兩種：Current Offset和Committed Offset萎馅。
Current Offset
Current Offset保存在Consumer客戶端中双戳，它表示Consumer希望收到的下一條消息的序號。它僅僅在poll()方法中使用糜芳。例如飒货，Consumer第一次調(diào)用poll()方法后收到了20條消息魄衅，那么Current Offset就被設(shè)置為20。這樣Consumer下一次調(diào)用poll()方法時塘辅，Kafka就知道應(yīng)該從序號為21的消息開始讀取晃虫。這樣就能夠保證每次Consumer poll消息時，都能夠收到不重復(fù)的消息扣墩。
Committed Offset
Committed Offset保存在Broker上哲银，它表示Consumer已經(jīng)確認(rèn)消費過的消息的序號。主要通過commitSync和commitAsync
API來操作呻惕。舉個例子荆责，Consumer通過poll() 方法收到20條消息后，此時Current Offset就是20亚脆，經(jīng)過一系列的邏輯處理后做院，并沒有調(diào)用consumer.commitAsync()或consumer.commitSync()來提交Committed Offset，那么此時Committed Offset依舊是0濒持。
Committed Offset主要用于Consumer Rebalance键耕。在Consumer Rebalance的過程中，一個partition被分配給了一個Consumer弥喉，那么這個Consumer該從什么位置開始消費消息呢郁竟？答案就是Committed Offset。另外由境，如果一個Consumer消費了5條消息（poll并且成功commitSync）之后宕機(jī)了棚亩，重新啟動之后它仍然能夠從第6條消息開始消費，因為Committed Offset已經(jīng)被Kafka記錄為5虏杰。
總結(jié)一下讥蟆，Current Offset是針對Consumer的poll過程的，它可以保證每次poll都返回不重復(fù)的消息纺阔；而Committed Offset是用于Consumer Rebalance過程的瘸彤，它能夠保證新的Consumer能夠從正確的位置開始消費一個partition，從而避免重復(fù)消費笛钝。
在Kafka 0.9前质况，Committed Offset信息保存在zookeeper的[consumers/{group}/offsets/{topic}/{partition}]目錄中（zookeeper其實并不適合進(jìn)行大批量的讀寫操作，尤其是寫操作）玻靡。而在0.9之后结榄，所有的offset信息都保存在了Broker上的一個名為__consumer_offsets的topic中。
Kafka集群中offset的管理都是由Group Coordinator中的Offset Manager完成的囤捻。
Group Coordinator
Group Coordinator是運行在Kafka集群中每一個Broker內(nèi)的一個進(jìn)程臼朗。它主要負(fù)責(zé)Consumer Group的管理，Offset位移管理以及Consumer Rebalance。
對于每一個Consumer Group视哑，Group Coordinator都會存儲以下信息：
- 訂閱的topics列表
- Consumer Group配置信息绣否，包括session timeout等
- 組中每個Consumer的元數(shù)據(jù)。包括主機(jī)名挡毅，consumer id
- 每個Group正在消費的topic partition的當(dāng)前offsets
- Partition的ownership元數(shù)據(jù)蒜撮，包括consumer消費的partitions映射關(guān)系
Consumer Group如何確定自己的coordinator是誰呢？ 簡單來說分為兩步：
- 確定Consumer Group offset信息將要寫入__consumers_offsets topic的哪個分區(qū)慷嗜。具體計算公式：
__consumers_offsets partition# = Math.abs(groupId.hashCode() % offsets.topic.num.partitions) //offsets.topic.num.partitions默認(rèn)值為50淀弹。
- 該分區(qū)leader所在的broker就是被選定的coordinator
Offset存儲模型
由于一個partition只能固定的交給一個消費者組中的一個消費者消費，因此Kafka保存offset時并不直接為每個消費者保存庆械，而是以groupid-topic-partition -> offset的方式保存薇溃。如圖所示：
Kafka在保存Offset的時候，實際上是將Consumer Group和partition對應(yīng)的offset以消息的方式保存在__consumers_offsets這個topic中缭乘。
__consumers_offsets默認(rèn)擁有50個partition沐序，可以通過
Math.abs(groupId.hashCode() % offsets.topic.num.partitions)
的方式來查詢某個Consumer Group的offset信息保存在__consumers_offsets的哪個partition中。下圖展示了__consumers_offsets中保存的offset消息的格式：
如圖所示堕绩，一條offset消息的格式為groupid-topic-partition -> offset策幼。因此consumer poll消息時，已知groupid和topic奴紧，又通過Coordinator分配partition的方式獲得了對應(yīng)的partition特姐，自然能夠通過Coordinator查找__consumers_offsets的方式獲得最新的offset了。
Offset查詢
前面我們已經(jīng)描述過offset的存儲模型黍氮，它是按照groupid-topic-partition -> offset的方式存儲的唐含。然而Kafka只提供了根據(jù)offset讀取消息的模型，并不支持根據(jù)key讀取消息的方式沫浆。那么Kafka是如何支持Offset的查詢呢捷枯？
答案就是Offsets Cache！专执！
如圖所示淮捆，Consumer提交offset時，Kafka Offset Manager會首先追加一條條新的commit消息到__consumers_offsets topic中本股，然后更新對應(yīng)的緩存攀痊。讀取offset時從緩存中讀取，而不是直接讀取__consumers_offsets這個topic拄显。
Log Compaction
我們已經(jīng)知道蚕苇，Kafka使用groupid-topic-partition -> offset*的消息格式，將Offset信息存儲在__consumers_offsets topic中凿叠。請看下面一個例子：
如圖，對于audit-consumer這個Consumer Group來說，上面的存儲了兩條具有相同key的記錄：PageViewEvent-0 -> 240和PageViewEvent-0 -> 323盒件。事實上蹬碧，這就是一種無用的冗余。因為對于一個partition來說炒刁，我們實際上只需要它當(dāng)前最新的Offsets恩沽。因此這條舊的PageViewEvent-0 -> 240記錄事實上是無用的。
為了消除這樣的過期數(shù)據(jù)翔始，Kafka為__consumers_offsets topic設(shè)置了Log Compaction功能罗心。Log Compaction意味著對于有相同key的的不同value值，只保留最后一個版本城瞎。如果應(yīng)用只關(guān)心key對應(yīng)的最新value值渤闷，可以開啟Kafka的Log Compaction功能，Kafka會定期將相同key的消息進(jìn)行合并脖镀，只保留最新的value值飒箭。
這張圖片生動的闡述了Log Compaction的過程：
下圖闡釋了__consumers_offsets topic中的數(shù)據(jù)在Log Compaction下的變化：
在新建topic時添加
log.cleanup.policy=compact參數(shù)就可以為topic開啟Log Compaction功能。
auto.offset.reset參數(shù)
auto.offset.reset表示如果Kafka中沒有存儲對應(yīng)的offset信息的話（有可能offset信息被刪除）蜒灰，消費者從何處開始消費消息弦蹂。它擁有三個可選值：
- earliest：從最早的offset開始消費
- latest：從最后的offset開始消費
- none：直接拋出exception給consumer
看一下下面兩個場景：
Consumer消費了5條消息后宕機(jī)了，重啟之后它讀取到對應(yīng)的partition的Committed Offset為5强窖，因此會直接從第6條消息開始讀取凸椿。此時完全依賴于Committed Offset機(jī)制，和
auto.offset.reset配置完全無關(guān)翅溺。新建了一個新的Group脑漫，并添加了一個Consumer，它訂閱了一個已經(jīng)存在的Topic未巫。此時Kafka中還沒有這個Consumer相應(yīng)的Offset信息窿撬，因此此時Kafka就會根據(jù)
auto.offset.reset配置來決定這個Consumer從何處開始消費消息。
