Kafka 與很多其它 MQ 不太一樣，Kafka 中的 Topic 具有分區(qū)（Partition）的概念，一個(gè) Topic 可以指定一個(gè)或多個(gè)分區(qū)，每個(gè)分區(qū)內(nèi)的數(shù)據(jù)都是有序的，發(fā)布的消息最終落到哪個(gè)分區(qū)是有規(guī)則的，默認(rèn)情況下隨機(jī)（但其實(shí)也不完全隨機(jī)，內(nèi)部會(huì)有一些機(jī)制），實(shí)際情況下我們可能會(huì)對(duì)消息指定一個(gè) Key， 相同 Key 的消息會(huì)落到一個(gè)分區(qū)上，不同的 Key 的消息也可能會(huì)落到相同的分區(qū)，分區(qū)與 Key 的關(guān)系的1對(duì)多，但 Key 與 分區(qū)的關(guān)系必須是1對(duì)1。

topic partition

消費(fèi)者組 (Consumer Group)

消費(fèi)者并不是獨(dú)立存在的，每一個(gè)消費(fèi)者實(shí)例必定屬于一個(gè)消費(fèi)組。

Consumer Group

消費(fèi)組主要說明：

• Consumer Group 需要設(shè)定一個(gè) group.id，Consumer Group 的唯一標(biāo)識(shí)；

• Consumer Group 下可以有一個(gè)或多個(gè) Consumer 實(shí)例，Consumer 實(shí)例可以是一個(gè)進(jìn)程，也可以是一個(gè)線程；

• Consumer Group 可以訂閱多個(gè) Topic，但一個(gè) Topic 下的一個(gè)分區(qū)只能分配給某個(gè) Consumer 實(shí)例消費(fèi)，不同的 Consumer Group 可以消費(fèi) Topic 下同一個(gè)分區(qū)的數(shù)據(jù)；

• Consumer Group 最多的 Consumer 實(shí)例個(gè)數(shù)不應(yīng)超過訂閱的 Topic 的分區(qū)數(shù)，因?yàn)橐粋€(gè)分區(qū)只能給一個(gè) Consumer 實(shí)例消費(fèi)，多出來的 Consumer 實(shí)例完全是浪費(fèi)；

在實(shí)際消費(fèi)數(shù)據(jù)時(shí)，可以根據(jù)消費(fèi)情況動(dòng)態(tài)對(duì) Topic 的分區(qū)數(shù)進(jìn)行調(diào)整，從而達(dá)到更高的效率。

消費(fèi)組的偏移重置設(shè)定（auto.offset.reset）

auto.offset.reset

largest(默認(rèn)) ，也可以使用 latest、end

當(dāng)各分區(qū)下有已提交的 offset 時(shí)，從提交的 offset 開始消費(fèi)；無提交的 offset 時(shí)，消費(fèi)新產(chǎn)生的該分區(qū)下的數(shù)據(jù)

注意 : 已提交的 offset 不是指 producer 發(fā)送的消息對(duì)應(yīng)的 offset ，而是指消費(fèi)者消費(fèi)的消息對(duì)應(yīng)的 offset 。

這到底是什么意思？這里對(duì)兩種情況進(jìn)行解釋一下：

1. 在沒有啟動(dòng) Consumer 實(shí)例的情況下，Producer 向某個(gè) Topic 發(fā)布了消息。之后當(dāng) Consumer 實(shí)例啟動(dòng)時(shí)是不會(huì)消費(fèi)之前發(fā)布的數(shù)據(jù)的，只有新發(fā)布的數(shù)據(jù)會(huì)被消費(fèi)；

2. 在 Consumer 實(shí)例啟動(dòng)的情況下，當(dāng) Producer 向某個(gè) Topic（n 個(gè)分區(qū)） 發(fā)布消息時(shí)，當(dāng) Consumer 實(shí)例掛掉時(shí)并沒有對(duì)所有分區(qū)有過消費(fèi)記錄，在掛掉期間，如果 Producer 發(fā)布的消息恰好落到?jīng)]有消費(fèi)記錄的分區(qū)，Consumer 實(shí)例重新啟動(dòng)后，這部分消息（落到?jīng)]有消費(fèi)記錄分區(qū)）將不會(huì)消費(fèi)，只有新發(fā)布的數(shù)據(jù)會(huì)被消費(fèi)；

smallest，也可以使用 earliest、beginning

當(dāng)各分區(qū)下有已提交的 offset 時(shí)，從提交的 offset 開始消費(fèi)；無提交的offset時(shí)，從頭開始消費(fèi)

error

Topic 各分區(qū)都存在已提交的 offset 時(shí)，從 offset 后開始消費(fèi)；只要有一個(gè)分區(qū)不存在已提交的 offset ，則拋出異常

如果沒有使用過 largest 或者 smallest 方式使 Topic 各分區(qū)下有已提交的 offset，設(shè)置成 error 必然是一直異常。

配置例子（C# 版）

var config = new Dictionary<string, object>
{
  { "group.id", "consumer-group" }, // 消費(fèi)組標(biāo)識(shí)
  { "bootstrap.servers", "127.0.0.1:9092" }, // broker地址，如果是集群，可以用 "," 分割
  { "auto.offset.reset", "smallest"} 
};

參考鏈接

Global configuration properties