Apache Kafka

Apache Kafka 是一種用於持久化事件串流的分散式追加日誌系統。生產者將記錄追加至主題分割區，代理程式複製並保留這些記錄，而消費者則依偏移量讀取記錄。因此，當多個獨立應用程式需要觀察相同的有序歷史，同時又不希望耦合其可用性或處理速率時， Kafka 非常適合。

叢集角色與中繼資料 #

Kafka 叢集包含兩種伺服器角色：

代理程式儲存分割區副本，並處理 produce 與 fetch 請求。
KRaft 控制器維護叢集中繼資料、代理程式成員關係、主題設定、副本放置與分割區領導權。

在開發叢集中，一個程序可以執行其中一個或兩個角色。生產環境叢集會分離控制器與代理程式角色，因為資料平面負載不應干擾中繼資料法定人數的決策。ZooKeeper 屬於 Kafka 的舊版中繼資料架構，在 Kafka 4.x 中已不存在。

客戶端接收的是啟動位址清單，而不是完整的路由表。中繼資料請求會識別目前負責各分割區的代理程式，之後 produce 與 fetch 請求會直接傳送至相關領導者。因此，每個公告的代理程式位址都必須能從客戶端網路解析並連線；向外部客戶端公布僅能在容器內使用的主機名稱，是常見的部署錯誤。

Kafka 在 TCP 之上使用具版本的二進位協定。org.apache.kafka:kafka-clients 工件提供 Java 生產者、消費者與管理 API。其他語言的客戶端實作相同協定。

主題、分割區與記錄 #

主題會分割成有序的分割區。每筆記錄可以包含：

一個鍵；
一個值；
一個時間戳記；以及
零個或多個標頭。

在協定邊界，鍵和值都是位元組陣列。序列化器會將應用程式物件轉換為這些位元組，反序列化器則執行反向操作。

代理程式會為追加至單一分割區的記錄指派單調遞增的偏移量。偏移量識別該分割區內的位置，不是全域有序的事件 ID。Kafka 僅保證單一分割區內的順序，不保證整個主題的順序。

當生產者未指定分割區時，分割器會使用記錄鍵與目前的主題中繼資料。因此，只要分割區數量維持不變，相等的非 null 鍵就會被路由至同一個分割區。這項特性支援對帳戶 ID 或裝置 ID 等鍵維持實體內排序。增加分割區數量可能改變鍵到分割區的對映，因此擴充分割區需要明確的排序與遷移分析。

每個分割區都以區段檔案的形式儲存在代理程式日誌目錄下。區段檔案包含二進位記錄批次以及偏移量和時間戳記索引；它們不是逐行排列的文字日誌。

保留與壓縮 #

消費不會刪除記錄。代理程式會根據主題的清理原則移除記錄：

cleanup.policy=delete 在達到設定的時間或大小邊界後移除舊區段。
cleanup.policy=compact 保留每個鍵的最新值。
cleanup.policy=compact,delete 同時套用兩項原則。

壓縮會保留偏移量，但不會產生密集的連續序列。因此，消費者可能會遇到偏移量之間的間隔。帶有 null 值的鍵控記錄稱為墓碑記錄；經過 delete.retention.ms 後，壓縮可能同時移除墓碑記錄及該鍵的舊值。

使用管理工具設定壓縮狀態主題：

bin/kafka-topics.sh --bootstrap-server broker:9092 \
  --create --topic customer-state \
  --partitions 12 --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.cleanable.dirty.ratio=0.5

壓縮是非同步的。應用程式不得假設替代值或墓碑記錄追加後，過時值會立即消失。

複寫與耐久性 #

一個分割區具有一個領導者和一個或多個副本。生產者和消費者通常與領導者通訊。追隨者會擷取領導者的日誌，而當領導權變更時，控制器會選出替代領導者。

多個獨立設定共同決定已確認寫入的耐久性：

replication.factor 決定儲存分割區的代理程式數量。
acks=all 使生產者等待領導者滿足同步副本需求。
min.insync.replicas 在同步中的副本太少時拒絕 acks=all 寫入。
允許非乾淨領導者選舉會讓過時副本成為領導者，因此可能遺失已確認的記錄。

例如，複寫因子為 3 且 min.insync.replicas=2 時，可以容忍一個副本無法使用，同時維持雙副本確認規則。但對寫入而言，無法容忍兩個副本無法使用。這項選擇使可用性與耐久性相互關聯。

副本放置必須跨越主機、機架或可用性區域等故障網域。將三個副本放在同一台實體主機上，無法提供主機故障容錯能力。

建立與檢查主題 #

主題數量與分割區數量是操作容量方面的決策，因此應明確建立重要主題，而不是依賴代理程式端的自動建立：

bin/kafka-topics.sh --bootstrap-server broker:9092 \
  --create --topic orders \
  --partitions 12 --replication-factor 3 \
  --config min.insync.replicas=2 \
  --config retention.ms=604800000

bin/kafka-topics.sh --bootstrap-server broker:9092 \
  --describe --topic orders

bin/kafka-configs.sh --bootstrap-server broker:9092 \
  --entity-type topics --entity-name orders --describe

Java Admin API 提供相同的控制平面：

java

import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;

Properties config = new Properties();
config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");

try (Admin admin = Admin.create(config)) {
    NewTopic orders = new NewTopic("orders", 12, (short) 3)
        .configs(Map.of(
            "min.insync.replicas", "2",
            "retention.ms", "604800000"
        ));
    admin.createTopics(List.of(orders)).all().get();
}

createTopics 是非同步的；等待傳回的 future 可以揭露授權、驗證與逾時失敗。

Java 客戶端依賴 #

生產者、消費者和 Admin 客戶端位於同一個工件中：

xml

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>${kafka.version}</version>
</dependency>

請從 Kafka 的相容性文件選擇客戶端版本。協定協商允許混用許多不同版本的客戶端與代理程式，但新的 API 和記錄格式仍需要最低代理程式版本。

產生記錄 #

KafkaProducer 是執行緒安全的。它會將記錄累積到各分割區批次中，壓縮每個批次，並非同步傳送請求。因此，長期執行的生產者比每筆記錄建立一個生產者有更好的效能。

java

import java.nio.charset.StandardCharsets;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

Properties config = new Properties();
config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class.getName());
config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class.getName());
config.put(ProducerConfig.ACKS_CONFIG, "all");
config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "zstd");
config.put(ProducerConfig.LINGER_MS_CONFIG, 5);

try (KafkaProducer<String, String> producer = new KafkaProducer<>(config)) {
    ProducerRecord<String, String> record =
        new ProducerRecord<>("orders", "customer-42", "{\"orderId\":123}");
    record.headers().add("schema-version",
        "1".getBytes(StandardCharsets.UTF_8));

    producer.send(record, (metadata, error) -> {
        if (error != null) {
            System.err.println("Produce failed: " + error);
            return;
        }
        System.out.printf("%s-%d@%d%n",
            metadata.topic(), metadata.partition(), metadata.offset());
    });
}

關閉生產者會清除尚未完成的傳送。應用程式應傳播或記錄回呼失敗；客戶端緩衝區接受記錄並不表示代理程式已確認該記錄。

重要的吞吐量與延遲控制項包括：

batch.size：一個分割區批次累積的最大位元組數；
linger.ms：部分填滿的批次可使用的時間；
compression.type：套用至整個批次的壓縮方式；
buffer.memory：生產者可提供給未傳送記錄的記憶體；
delivery.timeout.ms：完成傳送的上限；以及
max.request.size：客戶端接受的最大請求大小。

較大的批次可改善壓縮率與請求效率，但會消耗記憶體，並可能增加延遲。代理程式和主題的訊息大小限制也必須能接受產生的批次。

冪等性與重試 #

網路故障可能使生產者無法確定代理程式是否已追加請求。盲目重試可能因此產生重複。具冪等性的產生會附加生產者識別與序號，讓代理程式能拒絕生產者工作階段內的重複重試。

冪等性不會對兩個獨立的業務請求、由不同生產者執行個體傳送的記錄，或應用程式層級重播後重新傳送的記錄進行去重。穩定的業務 ID 和下游冪等性仍然是必要的。

消費記錄 #

KafkaConsumer 不是執行緒安全的。每個消費者應由一個應用程式執行緒擁有，負責呼叫 poll、處理分配回呼並提交偏移量。平行處理需要在已完成的工作與偏移量提交之間進行謹慎協調。

java

import java.time.Duration;
import java.util.List;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

Properties config = new Properties();
config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
config.put(ConsumerConfig.GROUP_ID_CONFIG, "order-indexer");
config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
    StringDeserializer.class.getName());
config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
    StringDeserializer.class.getName());
config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(config)) {
    consumer.subscribe(List.of("orders"));

    while (!Thread.currentThread().isInterrupted()) {
        ConsumerRecords<String, String> records =
            consumer.poll(Duration.ofSeconds(1));

        for (ConsumerRecord<String, String> record : records) {
            indexOrder(record.key(), record.value());
        }

        // Commits the next offsets after the complete batch succeeds.
        consumer.commitSync();
    }
}

此範例提供至少一次處理：如果在 indexOrder 之後、commitSync 之前發生故障，就會重播。因此，indexOrder 必須具備冪等性，或目的地必須依穩定的事件 ID 去重。

auto.offset.reset 只在群組沒有有效的已提交偏移量時套用。它不會強制既有群組重新讀取主題。若要刻意重播，請使用 seek、新的群組 ID 或 consumer-groups 重設操作。

消費者群組與重新平衡 #

具有相同 group.id 的消費者會形成一個群組。每個分割區在該群組中最多分配給一個作用中消費者：

消費者少於分割區時，部分消費者會擁有多個分割區；
數量相等時，每個消費者擁有一個分割區；以及
消費者多於分割區時，多出的消費者會處於閒置狀態。

不同群組會獨立接收主題歷史的副本。因此，詐欺偵測器和搜尋索引器應使用不同的群組 ID。

成員關係或訂閱變更時，分配會更新。協作式分配器可減少分割區移動，而靜態成員關係可以避免短暫重新啟動期間不必要的重新分配。不過，應用程式仍必須停止處理已撤銷的分割區，並且只提交已完成的偏移量。

使用以下命令檢查分配和延遲：

bin/kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --describe --group order-indexer

延遲量是 log-end-offset - committed-offset。它衡量未提交的距離，而不是處理延遲。低流量分割區可能具有較小的延遲量，但其中最早的未處理記錄可能已經很舊，因此應同時監控記錄延遲和以時間為基礎的服務目標。

max.poll.interval.ms 限制兩次呼叫 poll 之間的時間。同步處理大型批次可能超過此間隔，並使消費者從群組中移除。可使用 max.poll.records 限制批次，針對合理的慢速工作增加間隔，或在暫停分割區並只提交連續完成的偏移量時，將工作移至執行器。

傳遞語意 #

Kafka 應用程式可以實作三種結果：

至多一次： 在處理前前進或提交，接受故障後的遺失。
至少一次： 在提交前處理，接受重播與重複。
Kafka 精確一次處理： 使用 Kafka 交易，原子地提交產生的記錄和消費的偏移量。

Kafka 交易不會讓資料庫寫入、電子郵件或 HTTP 呼叫具備原子性。外部效果需要冪等鍵、 outbox/inbox 設計、去重或其他協調協定。

交易式 Consume-Transform-Produce #

每個運作中的生產者執行個體都需要穩定且唯一的 transactional.id。並行重複使用同一個 ID 會隔離舊的生產者，阻止兩個執行個體以相同交易擁有者身分提交。

java

import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

Properties consumerConfig = new Properties();
consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, "order-normalizer");
consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
consumerConfig.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
    StringDeserializer.class.getName());
consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
    StringDeserializer.class.getName());

Properties producerConfig = new Properties();
producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
producerConfig.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,
    "order-normalizer-instance-1");
producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class.getName());
producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
    StringSerializer.class.getName());

try (KafkaConsumer<String, String> consumer =
         new KafkaConsumer<>(consumerConfig);
     KafkaProducer<String, String> producer =
         new KafkaProducer<>(producerConfig)) {
    consumer.subscribe(List.of("orders"));
    producer.initTransactions();

    while (true) {
        var records = consumer.poll(Duration.ofSeconds(1));
        if (records.isEmpty()) {
            continue;
        }

        try {
            producer.beginTransaction();
            Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();

            for (var record : records) {
                producer.send(new ProducerRecord<>(
                    "normalized-orders", record.key(),
                    normalize(record.value())));
                offsets.put(
                    new TopicPartition(record.topic(), record.partition()),
                    new OffsetAndMetadata(record.offset() + 1)
                );
            }

            producer.sendOffsetsToTransaction(
                offsets, consumer.groupMetadata());
            producer.commitTransaction();
        } catch (ProducerFencedException fatal) {
            throw fatal;
        } catch (KafkaException retryableBatchFailure) {
            producer.abortTransaction();
        }
    }
}

必須隱藏已中止記錄的消費者應使用 isolation.level=read_committed。交易會增加協調成本和延遲，因此應在需要 Kafka 讀取、處理、寫入行為具備原子性時使用，而不是預設取代冪等處理。

Kafka Streams #

Kafka Streams 是可嵌入的 Java 函式庫，不是獨立的叢集服務。每個應用程式執行個體會加入消費者群組，將工作分割為工作，並儲存由變更日誌主題支援的本機狀態。

java

import java.util.Properties;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;

Properties config = new Properties();
config.put(StreamsConfig.APPLICATION_ID_CONFIG, "paid-order-counts");
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
    Serdes.String().getClass());
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
    Serdes.String().getClass());
config.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, "exactly_once_v2");

StreamsBuilder builder = new StreamsBuilder();
builder.<String, String>stream("orders")
    .filter((customerId, json) -> json.contains("\"status\":\"PAID\""))
    .groupByKey()
    .count()
    .toStream()
    .to("paid-orders-per-customer");

KafkaStreams streams = new KafkaStreams(builder.build(), config);
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
streams.start();

變更 application.id 會建立新的處理識別，並使用不同的消費者群組、內部主題和狀態目錄。應將其視為持久化的應用程式狀態，而非顯示名稱。

Kafka Connect 與複寫 #

Kafka Connect 執行可重複使用的來源和接收連接器。來源連接器從外部系統匯入記錄；接收連接器匯出 Kafka 記錄。分散式工作者將連接器設定、偏移量和狀態儲存在 Kafka 主題中，並在工作者之間重新平衡連接器工作。

Connect 簡化標準整合，但不會消除定義以下項目的必要性：

記錄鍵與分割；
結構描述相容性；
刪除和墓碑記錄語意；
重試和死信行為；
目的地冪等性；以及
憑證與網路邊界。

MirrorMaker 2 使用 Connect 在 Kafka 叢集之間複寫主題。複寫是非同步的，不會將兩個叢集變成一個同步共識系統。災害復原程序必須考慮複寫延遲、消費者偏移量轉換、主題設定和故障回復。

結構描述與演進 #

Kafka 儲存位元組，不會強制應用程式結構描述。Avro、Protobuf、JSON Schema 或其他編碼可以定義記錄結構。結構描述登錄中心可以將結構描述 ID 與承載內容關聯，並拒絕不相容的變更。

相容性規則應反映資料的重播方式：

新消費者可能遇到保留數月的舊記錄；
舊消費者可能在生產者變更期間仍然部署；
壓縮主題可能無限期保留舊值；以及
從封存還原可能重新引入舊結構描述。

新增具有定義預設值的選用欄位，比變更欄位意義或重新使用欄位名稱更容易演進。即使序列化格式允許機械式變更，事件名稱和欄位語意仍構成長期契約。

安全性 #

Kafka 將傳輸安全、驗證和授權分離：

TLS 加密流量，並向客戶端驗證代理程式。
mTLS 或 SASL 機制驗證客戶端。
ACL 授權對主題、群組、交易 ID 和叢集資源執行的操作。

可透過屬性提供 TLS 客戶端設定：

properties

security.protocol=SSL
ssl.truststore.location=/run/secrets/kafka.truststore.p12
ssl.truststore.password=${TRUSTSTORE_PASSWORD}
ssl.keystore.location=/run/secrets/client.keystore.p12
ssl.keystore.password=${KEYSTORE_PASSWORD}
ssl.key.password=${KEY_PASSWORD}

不要將真實密碼放在已提交的屬性檔案中。請透過部署平台的秘密機制注入密碼，並限制檔案系統存取權。

分割區與容量規劃 #

分割區數量限制消費者群組的平行度，但增加分割區並非沒有成本。每個分割區都會增加中繼資料、開啟的檔案、副本擷取工作、選舉和復原流量。可根據以下因素估算起始分割區數量：

尖峰入口和出口吞吐量；
最大有效消費者並行數；
每個代理程式的網路和磁碟吞吐量；
鍵偏斜；
複寫和復原目標；以及
在可以接受重新分割之前的預期成長量。

保留容量約為：

text

ingress bytes/second × retention seconds × replication factor

應為區段粒度、索引、壓縮、暫時性不平衡、重新指派和代理程式復原預留餘裕。能夠處理穩態流量、但無法在復原目標內重建故障代理程式的叢集，容量配置不足。

操作與故障診斷 #

至少監控：

離線和複寫不足的分割區；
控制器與代理程式的可用性；
produce 和 fetch 的延遲與錯誤率；
請求處理器和網路執行緒的飽和度；
磁碟使用量和磁碟延遲；
副本擷取延遲；
消費者偏移量延遲和記錄年齡；
交易與驗證失敗；以及
分割區不平衡和重新指派進度。

有用的診斷命令包括：

# Observe records without joining an application consumer group.
bin/kafka-console-consumer.sh --bootstrap-server broker:9092 \
  --topic orders --from-beginning \
  --property print.key=true --property print.offset=true

# Inspect one group's committed offsets and lag.
bin/kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --group order-indexer --describe

# Read broker API compatibility information.
bin/kafka-broker-api-versions.sh --bootstrap-server broker:9092

無法使用的領導者、不足的同步副本、無法連線的公告監聽器、授權失敗以及過載的代理程式，都可能在應用程式中表現為傳送逾時。在變更重試上限之前，應將客戶端例外與代理程式指標、分割區狀態和控制器事件相互關聯。

Apache Kafka

叢集角色與中繼資料 #

主題、分割區與記錄 #

保留與壓縮 #

複寫與耐久性 #

建立與檢查主題 #

Java 客戶端依賴 #

產生記錄 #

冪等性與重試 #

消費記錄 #

消費者群組與重新平衡 #

傳遞語意 #

交易式 Consume-Transform-Produce #

Kafka Streams #

Kafka Connect 與複寫 #

結構描述與演進 #

安全性 #

分割區與容量規劃 #

操作與故障診斷 #

參考資料 #

評論