Apache Kafka 的基本概念

基本概念

主題 Topic

topic 是 Kafka 最基礎的組織單位，類似於關聯式資料庫中的資料表。做為使用 kafka 的開發者，你最應該考慮的是和 topoc 相關的抽象。建立不同的 topic 儲存不同種類的 events，或者通過不同的 topic 儲存各種版本經過過濾、轉換後的同類 events。

topic 就是一連串 event 紀錄檔（相對 queue 來說，沒有索引，不允許從中間插入，只能通過追加的方式寫入）。對於紀錄檔來說，語意簡單，資料結構也簡單。第一，紀錄檔只支援追加寫入：當在紀錄檔中寫入訊息時，總是寫在末端。第二，讀取紀錄檔在紀錄檔中搜尋偏移量（offset），然後掃描有序的紀錄檔條目。第三，紀錄檔中一系列事件具有不可變性——覆水難收。紀錄檔簡單的語意，使其在 Kafka 中，進出 topic 時能保持較高的吞吐量，並且意味著複製 topic 將更加容易。

紀錄檔也是基本的持久化手段。傳統企業的訊息傳送系統有 topic 和 queue，該系統在傳送者和接受者之間暫時快取訊息。

因為 Kafka 的 topic 是紀錄檔，所以在 topic 中的資料不存在暫存的問題。每個 topic 都可以設定資料的有效期（或者固定大小），從幾秒鐘到幾年，甚至無限期。構成 Kafka topic 的紀錄檔做為檔案儲存在磁碟上。當你向 topic 中寫入事件時，該事件的永續性就像你將它寫入你信任的資料庫一樣。

簡單的紀錄檔、內容的不可變性是 Kafka 在現代資料基礎架構中，成功的關鍵——但這些只是開始。

Kafka 分割區

如果一個 topic 只能在一臺機器上存活，將極大的限制 Kafka 的可延伸性。Kafka 可以跨多臺機器管理多個 topic——畢竟，Kafka 是分散式系統——但沒有哪個主題能夠無限制的變大或者容納太多的讀寫。幸運的是，Kafka 賦予了我們將 topic 分割區的能力。

分割區將 topic 的單個紀錄檔檔案拆分為多個紀錄檔檔案，每一個紀錄檔檔案在 Kafka 叢集中，隔離的節點上都能存活。這樣一來，儲存訊息、寫入訊息、處理已有的訊息都能分離到叢集的節點中。

Kafka 如何分割區

如果將 topic 拆分到各個分割區，我們需要將訊息分配至分割區的方法。通常而言，如果訊息沒有 key，訊息將通過輪詢的方式寫入 topic 的各個分割區中。在這種情況下，所有分割區平均分配資料，但是我們不保留寫入的訊息順序。如果訊息有 key，將會通過 key 的雜湊值計算出目標分割區。這使得 Kafka 能保證相同的 key 寫入同一個分割區，並且有序。

例如，如果你生成一些和同一個客戶相關的事件，使用客戶的 ID 做為 key 將保證來自客戶的所有事件是有序的。但是這可能會導致，一個非常活躍的 key 建立一個又大、又活躍的分割區，但是在實踐中，這種風險很小，並且出現這種情況時，是可控的。大多數情況下，保持 key 的有序性是值得的。

Kafka Brokers

目前為止，我們討論了事件、主題、分割區，但是還沒有提到架構中的計算機。從物理基礎設施來說，Kafka 是由一個機器（稱為 broker）網路組成的。就目前的部署方式來說，這些 broker 可能不會部署在物理隔離的伺服器上，而是基於 pod 部署在容器中，而這些 pod 又是基於虛擬伺服器執行，虛擬伺服器基於某個物理資料中的核心執行。無論如何，這些 broker 部署後，它們之間互相獨立執行 Kafka broker 程序。每個 broker 持有一些分割區，並且處理請求以寫入新的事件到那些分割區中或者從那些分割區中讀取事件。broker 同時處理分割區之間的 replication。

Replication

如果我們只有一個 broker 儲存 partition，那麼 replication 將不工作。無論 broker 是部署在裸機上還是託管在容器中，它們以及它們潛在的儲存都可能受到宕機的影響，所以我們需要將分割區的資料拷貝至幾個 broker 中，從而保證資料的安全性。那些拷貝叫做從副本，反之，主分割區叫做主副本。當你生產資料到主分割區中時——一般而言，讀和寫都是通過主分割區完成的——主分割區和從分割區同時工作，複製那些新的寫入到從分割區中。

這個過程是自動的，而你可以在生產者中通過設定生成各種層級的持久化擔保，這個不是開發者基於 Kafka 構建系統時都需要考慮的工作。做為開發者，你只需要知道你的資料是安全的，如果叢集中的某個節點掛了，另一個節點將完全接手它的工作。

使用者端應用程式

現在，我們脫離 Kafka 叢集本身不談，看看使用 Kafka 的應用程式：生產者和消費者。這些使用者端應用程式，包含將訊息放入 topic 中，從 topic 中讀取訊息的程式碼。Kafka 平臺中的所有元件，除 broker 外，要麼是生產者，要麼是消費者，或者兩者兼備。生產和消費是你與叢集的互動方式。

Kafka 生產者

生產者庫的暴露的 API 相當輕量化：在 Java 中，有一個叫做 KafkaProducer 的類，你可以用其連線叢集。使用 Map 來設定引數，包括叢集中的幾個 broker 的地址，可用的安全設定，以及其他決定生產者網路行為的設定。還有一個類叫做 ProduceRecord，用於持有你要傳送到叢集中的鍵值對。

大致來說，這就是生產訊息所需的所有 API。從底層來說，該庫管理了連結池、網路緩衝、等待 broker 確認訊息、在必要的時候轉發訊息、以及一些開發者不需要關心的主機相關的 host 細節。

Kafka 消費者

消費者 API 原則上和生產者差不多。使用一個叫做 KafkaConsumer 的類連線叢集（使用 Map 設定叢集地址、安全性、以及其他引數）。然後使用該連結訂閱多個 topic。當 topic 中有可用的訊息時，使用者端以 ConsumerRecord 的格式接收訊息，並儲存在 ConsumerRecords 集合中。一個 ConsumerRecod 物件代表一條 Kafka 訊息的鍵值對。

KafkaConsumer 管理連結池以及網路協定，就和 KafkaProducer 一樣，但是在讀取方面，除了網路管道之外，還有更多的內容。首當其衝的是，Kafka 與其他訊息佇列不同，只是讀取訊息，不銷燬訊息；其他消費者仍然可以消費那些訊息。事實上，在 Kafka 中，多個消費者從一個 topic 中消費資訊非常正常。這個小事實對結合 Kafka 的軟體架構來說，產生了極大的影響。

同時，消費者需要處理以下這種情況，某個主題的消費速率，結合處理單條訊息的計算成本後，單個範例無法承受。也就是說，需要擴充套件消費者。在 Kafka 中，擴充套件消費者組或多或少是自動的。

Kafka 元件、生態系統

如果你的系統只是隨著生產者（寫事件）和消費者（讀事件）的不斷增長，通過 broker 管理分割區、複製 topic，那麼你的系統將是一個非常有用的系統。相比之下，Kafka 社群的經驗表明，某些模式顯現出來，並促使你和你的組員圍繞核心 Kafka 重複的構建相同的功能。

最終，你將構建一些公共的應用程式層，從而避免重複的無差別任務。這些程式碼的能力非常重要，但是和你的業務完全無關。並且對你的使用者不產生直接價值。這些是基礎設施，並且應該由社群或者基礎設施供應商提供。

寫這些程式碼可能很有誘惑力，但是你不應該這麼做。這類基礎設施程式碼有：Kafka Connect、Confulent Schema Registry、Kafka Streams、ksqlDB。接下來，我們將逐一進行講解。

Kafk Connect

在儲存和檢索的世界裡，不都是使用 Kafka。有時候，你可能想把儲存在那些系統裡的資料存入 Kafka topic 中，有時候，你想把 Kafka topic 裡的資料存入那些系統中。做為 Apache Kafka 的整合 API，Kafka Connect 負責這些能力。

Kafka Connect 做了哪些事情？

一方面，Kafka Connect 是可插拔聯結器中的一個生態系統，另一方面，它也是一個使用者端應用程式。做為使用者端應用程式，Connect 是一個直接在硬體上執行的伺服器端程序，獨立於 Kafka broker。Connect 可延伸並且容錯，這表明你不僅可以執行單一的 Connect worker，還可以執行 Connect worker 叢集，這些 worker 共用從外部系統進出 Kafka 的負載。 Kafka Connect 同時抽象了編碼，只需要使用 JSON 設定即可執行。以下從 Kafka 到 Elasticsearch 的流資料設定範例：

{
  "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
  "topics"         : "my_topic",
  "connection.url" : "http://elasticsearch:9200",
  "type.name"      : "_doc",
  "key.ignore"     : "true",
  "schema.ignore"  : "true"
}

Kafka Connect 的優勢

Kafka Connect 最主要的優勢是聯結器的強大生態。將資料寫入雲端儲存、ES、或者關聯式資料庫中的程式碼，在不同的業務之間，不會產生很大的變化。同樣，從關聯式資料庫、Salesforce、或者遺留 HDFS 檔案系統讀取資料也是相似的。你可以自己編寫這部分程式碼，但是花費的時間並不會為你的客戶帶來獨特的價值，也不會讓你的業務更有競爭力。

Kafka Streams

在一個持續更新、整合了 Kafka 的應用程式中，消費者會變得越來越複雜。剛開始的時候，可能只是一個簡單的無狀態轉換器（例如，個人資訊脫敏、轉換訊息格式），不久後，整合進複雜的聚合、填充資料等操作。如果你回顧我們上面提到的消費者程式碼，就會發現 API 不支援那些操作：你將要構建很多框架程式碼來處理時間視窗、延遲訊息、查詢資料表、通過 Key 做整合，等等。等你實現那些程式碼後，你會發現像聚合、填充資料這類操作通常是有狀態的。

「狀態」需要儲存在程式的堆中，這就意味著它的容錯債務。如果你的流處理應用程式下線了，應用程式中的狀態也就丟失了，除非你設計了一張表來儲存狀態。在可延伸的情況下，編寫、偵錯這類事情並不簡單，同時也不會讓你的使用者受益。這就是為什麼 Apache Kafka 提供流處理 API 的原因。也是為什麼需要 Kafka Streams 的原因。

Kafka Streams 是什麼

Kafka Streams 是一個 Java API，在存取流處理計算原語時，為你提供便利性：過濾、分組、聚合、連線，等等，也不需要你基於消費者 API 編寫框架程式碼。同時支援流處理計算產生的大量狀態。如果你對一個吞吐量較高的 topic ，通過某個欄位做事件分組，然後每個小時對分組資料做一次彙總，那麼你可能需要很大的記憶體。

確實，對於資料量較大的 topic，複雜的流處理拓撲圖來說，不難想象，你必須要部署一個叢集來共用流處理負載，就和消費組一樣。Steams API 為你解決了分散式狀態的問題：它將狀態持久化至本地磁碟和 Kafka 叢集的 topic 中，並且在流處理叢集中新增或移除流處理節點時，在多節點間自動分配狀態。

在一個典型的微服務中，流處理是其他功能的附加產物。例如，發貨通知服務可能會將產品資訊變更紀錄檔（包括客戶記錄）結合起來，生成發貨通知物件，其他服務將其轉換為電子郵件或者簡訊。但是，當為移動 App 或者 Web 前端展示發貨狀態時，發貨通知服務也有義務通過 REST API 封裝同步關鍵字檢索介面。

該服務是基於事件響應的服務——在這種情況下，連線（join） 3 個流，可能還要基於連線（join）結果做視窗計算——但是它也提供了基於 REST 介面提供的 HTTP 服務，可能基於 Spring 框架或者 Micronaut，或者其他常用的 Java API。鑑於 Kafka Streams 是一個只做流處理的 Java 庫，而不是一組專門做流處理的基礎元件，所以，使用其他框架實現 REST 介面，以及複雜的、可延伸的、容錯的流處理非常容易。