Kafka消費者配置與使用詳解

– 20 分鐘閱讀

Apache Kafka消費者是串流處理架構中負責讀取和處理訊息的核心元件。與傳統訊息佇列的消費者不同，Kafka消費者採用拉取模式主動從Broker獲取資料，這種設計賦予了消費者更大的控制權，可以根據自身的處理能力調整消費速率。Kafka消費者的配置參數眾多且相互關聯，正確理解和設定這些參數對於建構高效、可靠的串流處理應用程式至關重要。本文將深入探討Kafka消費者的運作機制、關鍵配置參數、分區分配策略以及效能調校技巧，幫助讀者全面掌握Kafka消費者的使用方法，建構穩定高效的訊息消費系統。

@startuml
!define PLANTUML_FORMAT svg
!theme _none_

skinparam dpi auto
skinparam shadowing false
skinparam linetype ortho
skinparam roundcorner 5
skinparam defaultFontName "Microsoft JhengHei UI"
skinparam defaultFontSize 16
skinparam minClassWidth 100

rectangle "Kafka 消費者架構" as arch {
    rectangle "消費者群組" as group {
        component "消費者 1" as c1
        component "消費者 2" as c2
        component "消費者 3" as c3
    }

    rectangle "群組協調器" as coordinator {
        component "成員管理" as member
        component "分區分配" as assign
        component "偏移量管理" as offset
    }

    rectangle "Kafka 叢集" as cluster {
        component "主題 A" as topicA
        component "主題 B" as topicB
        component "內部主題" as internal
    }

    rectangle "消費者配置" as config {
        component "連接配置" as conn
        component "擷取配置" as fetch
        component "心跳配置" as heartbeat
        component "提交配置" as commit
    }
}

c1 --> coordinator
c2 --> coordinator
c3 --> coordinator
coordinator --> cluster
member --> assign
assign --> offset
config --> group

@enduml

Kafka消費者群組是實現消費者負載平衡和容錯的核心機制。同一個消費者群組中的多個消費者會協同工作，每個分區只會被群組中的一個消費者處理，這確保了訊息不會被重複消費。當消費者加入或離開群組時，Kafka會觸發重新平衡操作，重新分配分區給現有的消費者。這種設計使得應用程式可以透過簡單地增加或減少消費者數量來實現水平擴展。群組協調器負責管理消費者群組的成員資格、分區分配以及偏移量提交，它是整個消費者群組運作的核心。

消費者與群組協調器之間的通訊透過心跳機制維持。消費者會定期向協調器發送心跳，表明自己仍然存活且正在處理訊息。如果協調器在指定時間內沒有收到某個消費者的心跳，就會認為該消費者已經失效，並觸發重新平衡。這個機制確保了消費者群組能夠快速偵測並處理失效的消費者，維持系統的可用性。

建立Kafka消費者是使用Consumer API的第一步。與Producer類似，消費者需要透過配置屬性來定義其行為。最基本的配置包括Bootstrap伺服器地址、鍵和值的反序列化器，以及消費者群組ID。這些配置決定了消費者如何連接到Kafka叢集、如何解析訊息內容，以及屬於哪個消費者群組。

// Kafka 消費者建立與基本配置範例
// 這段程式碼展示了建立 KafkaConsumer 實例的完整流程

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;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class BasicKafkaConsumer {

    public static void main(String[] args) {
        // 建立配置屬性物件
        // Properties 物件用於儲存所有消費者配置參數
        Properties props = new Properties();

        // 設定 Bootstrap 伺服器
        // 指定 Kafka 叢集的連接地址
        // 可以設定多個 Broker 地址，以逗號分隔
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
            "broker1:9092,broker2:9092,broker3:9092");

        // 設定消費者群組 ID
        // 同一群組的消費者會協同消費主題中的訊息
        // 每個分區只會被群組中的一個消費者處理
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-group");

        // 設定鍵的反序列化器
        // 指定如何將位元組陣列轉換為鍵物件
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
            StringDeserializer.class.getName());

        // 設定值的反序列化器
        // 指定如何將位元組陣列轉換為值物件
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
            StringDeserializer.class.getName());

        // 設定自動偏移量重置策略
        // earliest: 從最早的訊息開始消費
        // latest: 從最新的訊息開始消費
        // none: 如果沒有有效偏移量則拋出異常
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 設定是否自動提交偏移量
        // false 表示手動控制偏移量提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        // 設定客戶端 ID
        // 用於日誌記錄和監控識別
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, "order-consumer-1");

        // 建立 KafkaConsumer 實例
        // 泛型參數指定鍵和值的類型
        KafkaConsumer<String, String> consumer =
            new KafkaConsumer<>(props);

        // 訂閱主題
        // 可以訂閱一個或多個主題
        consumer.subscribe(Collections.singletonList("orders"));

        // 輸出消費者配置資訊
        System.out.println("消費者已建立並訂閱主題 'orders'");
        System.out.println("消費者群組: " + props.get(ConsumerConfig.GROUP_ID_CONFIG));
        System.out.println("Bootstrap 伺服器: " +
            props.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG));

        // 關閉消費者時釋放資源
        // 實際應用中應該在適當的時機調用
        // consumer.close();
    }
}

訂閱主題是消費者開始接收訊息的前提。Kafka提供了兩種訂閱方式：明確指定主題列表和使用正規表示式模式匹配。明確指定主題是最常見的方式，適用於消費者需要處理特定主題的場景。正規表示式訂閱則更加靈活，當新的主題被建立且名稱符合模式時，消費者會自動開始消費該主題。這種方式特別適合需要動態處理多個相關主題的應用程式。

// 主題訂閱範例
// 展示不同的主題訂閱方式

import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.regex.Pattern;

public class TopicSubscriptionExamples {

    public void subscribeToTopics(KafkaConsumer<String, String> consumer) {
        // 方式一：訂閱單一主題
        // 使用 Collections.singletonList 建立單元素列表
        consumer.subscribe(java.util.Collections.singletonList("user-events"));

        // 方式二：訂閱多個主題
        // 使用 Arrays.asList 建立多元素列表
        consumer.subscribe(Arrays.asList(
            "user-events",
            "order-events",
            "payment-events"
        ));

        // 方式三：使用正規表示式訂閱
        // 匹配所有以 "events-" 開頭的主題
        // 當新的符合模式的主題被建立時，消費者會自動訂閱
        consumer.subscribe(Pattern.compile("events-.*"));

        // 方式四：訂閱特定環境的主題
        // 例如只訂閱生產環境的主題
        consumer.subscribe(Pattern.compile("prod\\..*\\.events"));
    }

    // 使用 ConsumerRebalanceListener 監聽重新平衡事件
    // 這在需要清理或初始化資源時非常有用
    public void subscribeWithListener(KafkaConsumer<String, String> consumer) {
        consumer.subscribe(
            Arrays.asList("orders", "payments"),
            new org.apache.kafka.clients.consumer.ConsumerRebalanceListener() {

                @Override
                public void onPartitionsRevoked(
                    java.util.Collection<org.apache.kafka.common.TopicPartition> partitions
                ) {
                    // 分區被撤銷時調用
                    // 可以在這裡提交偏移量或清理資源
                    System.out.println("分區被撤銷: " + partitions);

                    // 提交當前偏移量
                    consumer.commitSync();
                }

                @Override
                public void onPartitionsAssigned(
                    java.util.Collection<org.apache.kafka.common.TopicPartition> partitions
                ) {
                    // 分區被分配時調用
                    // 可以在這裡初始化資源或載入狀態
                    System.out.println("分區已分配: " + partitions);

                    // 可以在這裡設定消費起始位置
                    // for (TopicPartition partition : partitions) {
                    //     consumer.seek(partition, getStoredOffset(partition));
                    // }
                }
            }
        );
    }
}

輪詢迴圈是Kafka消費者的核心運作機制。消費者透過反覆調用poll方法從Broker獲取訊息批次，然後處理這些訊息。poll方法不僅負責獲取訊息，還處理了許多背景工作，包括發現新的分區、加入消費者群組、發送心跳以及處理重新平衡。因此，確保poll方法被定期調用是維持消費者正常運作的關鍵。

// 完整的消費者輪詢迴圈範例
// 展示生產環境中的最佳實踐

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;

public class ConsumerPollingLoop {

    // 用於控制消費者關閉的標誌
    private final AtomicBoolean running = new AtomicBoolean(true);
    private KafkaConsumer<String, String> consumer;

    public void consume() {
        // 建立消費者配置
        Properties props = createConsumerProperties();

        // 建立消費者實例
        consumer = new KafkaConsumer<>(props);

        // 訂閱主題
        consumer.subscribe(Arrays.asList("events", "notifications"));

        // 設定輪詢超時時間
        // 這個值決定了 poll 方法在沒有訊息時的等待時間
        Duration pollTimeout = Duration.ofMillis(100);

        // 用於追蹤處理統計
        long totalProcessed = 0;
        long lastLogTime = System.currentTimeMillis();

        try {
            // 主輪詢迴圈
            // 持續運行直到收到關閉信號
            while (running.get()) {
                // 調用 poll 方法獲取訊息批次
                // poll 方法會阻塞指定的超時時間
                ConsumerRecords<String, String> records = consumer.poll(pollTimeout);

                // 檢查是否有訊息
                if (records.isEmpty()) {
                    continue;
                }

                // 處理每個分區的訊息
                // 按分區處理可以更好地管理偏移量提交
                for (org.apache.kafka.common.TopicPartition partition : records.partitions()) {
                    // 獲取該分區的所有記錄
                    List<ConsumerRecord<String, String>> partitionRecords =
                        records.records(partition);

                    // 處理該分區的每條記錄
                    for (ConsumerRecord<String, String> record : partitionRecords) {
                        // 處理單條訊息
                        processRecord(record);
                        totalProcessed++;
                    }

                    // 獲取該分區的最後偏移量
                    long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();

                    // 為該分區提交偏移量
                    // 提交的偏移量是下一條要讀取的訊息的偏移量
                    consumer.commitSync(Collections.singletonMap(
                        partition,
                        new OffsetAndMetadata(lastOffset + 1)
                    ));
                }

                // 定期輸出處理統計
                long currentTime = System.currentTimeMillis();
                if (currentTime - lastLogTime > 60000) {
                    System.out.println("已處理訊息數: " + totalProcessed);
                    lastLogTime = currentTime;
                }
            }
        } catch (WakeupException e) {
            // 當調用 consumer.wakeup() 時會拋出此異常
            // 用於安全地中斷 poll 方法
            if (running.get()) {
                throw e;
            }
            // 如果是預期的關閉，忽略此異常
        } finally {
            // 關閉消費者並釋放資源
            consumer.close();
            System.out.println("消費者已關閉，總共處理了 " + totalProcessed + " 條訊息");
        }
    }

    // 處理單條訊息的方法
    private void processRecord(ConsumerRecord<String, String> record) {
        // 輸出訊息詳細資訊
        System.out.printf(
            "主題: %s, 分區: %d, 偏移量: %d, 鍵: %s, 值: %s, 時間戳: %d%n",
            record.topic(),
            record.partition(),
            record.offset(),
            record.key(),
            record.value(),
            record.timestamp()
        );

        // 在這裡實現實際的業務邏輯
        // 例如：
        // - 將訊息寫入資料庫
        // - 調用外部服務
        // - 更新快取
        // - 聚合計算
    }

    // 建立消費者配置的方法
    private Properties createConsumerProperties() {
        Properties props = new Properties();

        // 基本連接配置
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "event-processor");
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, "event-processor-1");

        // 反序列化器配置
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
            StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
            StringDeserializer.class.getName());

        // 偏移量配置
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        // 擷取配置
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "1024");
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "500");
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "500");

        // 心跳配置
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "10000");
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "300000");

        return props;
    }

    // 安全關閉消費者的方法
    public void shutdown() {
        running.set(false);
        // 使用 wakeup 方法中斷 poll 調用
        if (consumer != null) {
            consumer.wakeup();
        }
    }
}

執行緒安全是使用Kafka消費者時必須注意的重要議題。KafkaConsumer不是執行緒安全的，除了wakeup方法外，所有方法都應該從同一個執行緒調用。如果需要在多執行緒環境中使用消費者，最簡單的方式是為每個執行緒建立一個獨立的消費者實例。這種模式稱為一消費者一執行緒模式，是Kafka官方推薦的做法。

// 多執行緒消費者管理範例
// 展示如何安全地管理多個消費者執行緒

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

public class MultiThreadedConsumerManager {

    // 消費者執行緒數量
    private final int numConsumers;
    // 執行緒池
    private final ExecutorService executor;
    // 消費者執行緒列表
    private final List<ConsumerRunnable> consumers;
    // 運行狀態標誌
    private final AtomicBoolean running = new AtomicBoolean(true);

    public MultiThreadedConsumerManager(int numConsumers) {
        this.numConsumers = numConsumers;
        // 建立固定大小的執行緒池
        this.executor = Executors.newFixedThreadPool(numConsumers);
        this.consumers = new ArrayList<>();
    }

    // 啟動所有消費者
    public void start(String topic, String groupId) {
        // 為每個執行緒建立一個消費者
        for (int i = 0; i < numConsumers; i++) {
            ConsumerRunnable consumer = new ConsumerRunnable(
                i,
                topic,
                groupId,
                running
            );
            consumers.add(consumer);
            executor.submit(consumer);
        }

        System.out.println("已啟動 " + numConsumers + " 個消費者執行緒");
    }

    // 關閉所有消費者
    public void shutdown() {
        // 設定關閉標誌
        running.set(false);

        // 喚醒所有消費者
        for (ConsumerRunnable consumer : consumers) {
            consumer.wakeup();
        }

        // 關閉執行緒池
        executor.shutdown();

        try {
            // 等待執行緒結束
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }

        System.out.println("所有消費者已關閉");
    }

    // 消費者執行緒類別
    private static class ConsumerRunnable implements Runnable {
        private final int id;
        private final String topic;
        private final String groupId;
        private final AtomicBoolean running;
        private KafkaConsumer<String, String> consumer;

        public ConsumerRunnable(
            int id,
            String topic,
            String groupId,
            AtomicBoolean running
        ) {
            this.id = id;
            this.topic = topic;
            this.groupId = groupId;
            this.running = running;
        }

        @Override
        public void run() {
            // 建立消費者配置
            Properties props = new Properties();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
            props.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer-" + id);
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

            // 建立消費者
            consumer = new KafkaConsumer<>(props);

            try {
                // 訂閱主題
                consumer.subscribe(Collections.singletonList(topic));

                System.out.println("消費者 " + id + " 已啟動");

                // 輪詢迴圈
                while (running.get()) {
                    ConsumerRecords<String, String> records =
                        consumer.poll(Duration.ofMillis(100));

                    for (ConsumerRecord<String, String> record : records) {
                        // 處理訊息
                        System.out.printf(
                            "消費者 %d: 主題=%s, 分區=%d, 偏移量=%d%n",
                            id,
                            record.topic(),
                            record.partition(),
                            record.offset()
                        );
                    }

                    // 同步提交偏移量
                    if (!records.isEmpty()) {
                        consumer.commitSync();
                    }
                }
            } catch (org.apache.kafka.common.errors.WakeupException e) {
                if (running.get()) {
                    throw e;
                }
            } finally {
                consumer.close();
                System.out.println("消費者 " + id + " 已關閉");
            }
        }

        // 喚醒消費者以便關閉
        public void wakeup() {
            if (consumer != null) {
                consumer.wakeup();
            }
        }
    }

    // 使用範例
    public static void main(String[] args) throws InterruptedException {
        // 建立消費者管理器
        MultiThreadedConsumerManager manager = new MultiThreadedConsumerManager(3);

        // 啟動消費者
        manager.start("events", "event-consumers");

        // 運行一段時間
        Thread.sleep(60000);

        // 關閉消費者
        manager.shutdown();
    }
}

資料擷取配置參數控制消費者如何從Broker獲取訊息。fetch.min.bytes指定消費者希望從Broker獲取的最小資料量，如果可用資料量小於這個值，Broker會等待更多資料到達。這個配置可以減少網路往返次數，但會增加延遲。fetch.max.wait.ms則設定Broker在等待足夠資料時的最大等待時間，它與fetch.min.bytes配合使用，確保消費者不會等待太久。

max.poll.records參數控制單次poll調用返回的最大記錄數。這個值應該根據應用程式的處理能力來設定，太大的值可能導致處理時間過長，超過max.poll.interval.ms的限制，從而觸發重新平衡。max.partition.fetch.bytes則限制每個分區返回的最大資料量，用於控制記憶體使用。

// 資料擷取配置最佳實踐範例
// 展示如何根據不同場景配置擷取參數

import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.util.Properties;

public class FetchConfigurationExamples {

    // 低延遲場景配置
    // 適用於需要即時處理訊息的應用
    public Properties createLowLatencyConfig() {
        Properties props = new Properties();

        // 基本配置
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "low-latency-group");

        // 設定較小的最小擷取位元組數
        // 這樣即使只有少量資料也會立即返回
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "1");

        // 設定較短的最大等待時間
        // 減少在沒有足夠資料時的等待時間
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "100");

        // 設定較小的每次輪詢記錄數
        // 確保快速處理並返回
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "100");

        // 設定適中的每分區擷取限制
        props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG,
            String.valueOf(1024 * 1024)); // 1MB

        return props;
    }

    // 高吞吐量場景配置
    // 適用於批次處理或資料管道應用
    public Properties createHighThroughputConfig() {
        Properties props = new Properties();

        // 基本配置
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "high-throughput-group");

        // 設定較大的最小擷取位元組數
        // 等待累積足夠資料後再返回，減少網路往返
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG,
            String.valueOf(1024 * 1024)); // 1MB

        // 設定較長的最大等待時間
        // 允許更多時間來累積資料
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "500");

        // 設定較大的最大擷取位元組數
        props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG,
            String.valueOf(50 * 1024 * 1024)); // 50MB

        // 設定較大的每次輪詢記錄數
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1000");

        // 設定較大的每分區擷取限制
        props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG,
            String.valueOf(10 * 1024 * 1024)); // 10MB

        // 設定較長的最大輪詢間隔
        // 允許更多時間處理大批次的訊息
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "600000"); // 10分鐘

        return props;
    }

    // 記憶體受限場景配置
    // 適用於容器環境或記憶體有限的系統
    public Properties createMemoryConstrainedConfig() {
        Properties props = new Properties();

        // 基本配置
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "memory-constrained-group");

        // 限制總擷取大小
        props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG,
            String.valueOf(10 * 1024 * 1024)); // 10MB

        // 限制每分區擷取大小
        props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG,
            String.valueOf(256 * 1024)); // 256KB

        // 限制每次輪詢記錄數
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "50");

        // 設定較小的接收緩衝區
        props.put(ConsumerConfig.RECEIVE_BUFFER_CONFIG,
            String.valueOf(32 * 1024)); // 32KB

        return props;
    }
}

心跳和會話配置對消費者群組的穩定性至關重要。session.timeout.ms定義了消費者在未發送心跳情況下被視為失效的最大時間。heartbeat.interval.ms則控制心跳發送的頻率，通常應設為session.timeout.ms的三分之一。這兩個參數需要根據網路延遲和處理時間來調整，過短的超時可能導致不必要的重新平衡，過長則可能延遲故障檢測。

max.poll.interval.ms是另一個重要的配置，它定義了兩次poll調用之間的最大允許間隔。如果消費者在這個時間內沒有調用poll，協調器會認為它已經失效並觸發重新平衡。這個配置應該設定為足夠長，以允許處理一個完整的訊息批次，但又不能太長，否則會延遲故障檢測。

// 心跳與會話配置範例
// 展示如何配置消費者的存活檢測機制

import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.util.Properties;

public class HeartbeatSessionConfig {

    // 快速故障檢測配置
    // 適用於需要快速重新平衡的場景
    public Properties createFastFailoverConfig() {
        Properties props = new Properties();

        // 設定較短的會話超時
        // 消費者在此時間內未發送心跳將被視為失效
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "10000"); // 10秒

        // 設定心跳間隔為會話超時的三分之一
        // 這是 Kafka 推薦的比例
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "3000"); // 3秒

        // 設定適中的最大輪詢間隔
        // 需要確保處理時間不會超過此值
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "60000"); // 1分鐘

        return props;
    }

    // 穩定性優先配置
    // 適用於處理時間較長但不希望頻繁重新平衡的場景
    public Properties createStabilityFirstConfig() {
        Properties props = new Properties();

        // 設定較長的會話超時
        // 減少因網路延遲導致的誤判
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "45000"); // 45秒

        // 心跳間隔
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "15000"); // 15秒

        // 設定較長的最大輪詢間隔
        // 允許更長的處理時間
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "600000"); // 10分鐘

        return props;
    }

    // 跨區域消費配置
    // 適用於消費者和 Broker 位於不同資料中心的場景
    public Properties createCrossRegionConfig() {
        Properties props = new Properties();

        // 設定更長的會話超時以適應網路延遲
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "60000"); // 1分鐘

        // 設定較長的心跳間隔
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "20000"); // 20秒

        // 設定更長的請求超時
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, "60000"); // 1分鐘

        // 設定更大的接收和發送緩衝區
        // 減少網路往返的影響
        props.put(ConsumerConfig.RECEIVE_BUFFER_CONFIG,
            String.valueOf(512 * 1024)); // 512KB
        props.put(ConsumerConfig.SEND_BUFFER_CONFIG,
            String.valueOf(512 * 1024)); // 512KB

        return props;
    }
}

偏移量重置策略決定了消費者在沒有有效提交偏移量時從哪裡開始消費。auto.offset.reset參數有三個可選值：latest表示從最新的訊息開始，適用於只關心新訊息的場景；earliest表示從最早的訊息開始，適用於需要處理歷史訊息的場景；none則會在沒有有效偏移量時拋出異常，適用於需要明確控制起始位置的場景。

自動提交偏移量配置控制消費者是否自動提交偏移量。當enable.auto.commit設為true時，消費者會在poll方法中自動提交偏移量，提交間隔由auto.commit.interval.ms控制。自動提交雖然簡單，但可能導致重複消費或訊息遺失。對於需要精確一次處理語義的應用程式，應該禁用自動提交並手動管理偏移量。

// 偏移量管理範例
// 展示不同的偏移量管理策略

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.*;

public class OffsetManagementExamples {

    // 同步提交偏移量
    // 最簡單但可能阻塞的方式
    public void syncCommit(KafkaConsumer<String, String> consumer) {
        while (true) {
            ConsumerRecords<String, String> records =
                consumer.poll(Duration.ofMillis(100));

            for (ConsumerRecord<String, String> record : records) {
                // 處理訊息
                processRecord(record);
            }

            // 同步提交所有已消費訊息的偏移量
            // 這會阻塞直到 Broker 確認
            try {
                consumer.commitSync();
            } catch (CommitFailedException e) {
                // 處理提交失敗
                System.err.println("偏移量提交失敗: " + e.getMessage());
            }
        }
    }

    // 非同步提交偏移量
    // 不會阻塞，但可能在失敗時丟失偏移量
    public void asyncCommit(KafkaConsumer<String, String> consumer) {
        while (true) {
            ConsumerRecords<String, String> records =
                consumer.poll(Duration.ofMillis(100));

            for (ConsumerRecord<String, String> record : records) {
                processRecord(record);
            }

            // 非同步提交偏移量
            // 使用回調處理結果
            consumer.commitAsync(
                (offsets, exception) -> {
                    if (exception != null) {
                        System.err.println("非同步提交失敗: " + exception.getMessage());
                    } else {
                        System.out.println("已提交偏移量: " + offsets);
                    }
                }
            );
        }
    }

    // 混合提交策略
    // 正常情況使用非同步提交，關閉時使用同步提交
    public void mixedCommit(KafkaConsumer<String, String> consumer) {
        try {
            while (true) {
                ConsumerRecords<String, String> records =
                    consumer.poll(Duration.ofMillis(100));

                for (ConsumerRecord<String, String> record : records) {
                    processRecord(record);
                }

                // 正常情況使用非同步提交以提高效能
                consumer.commitAsync();
            }
        } finally {
            try {
                // 關閉前使用同步提交確保偏移量被保存
                consumer.commitSync();
            } finally {
                consumer.close();
            }
        }
    }

    // 精確控制偏移量提交
    // 在處理每條訊息後提交該訊息的偏移量
    public void preciseCommit(KafkaConsumer<String, String> consumer) {
        while (true) {
            ConsumerRecords<String, String> records =
                consumer.poll(Duration.ofMillis(100));

            for (ConsumerRecord<String, String> record : records) {
                // 處理訊息
                processRecord(record);

                // 建立該訊息的偏移量元資料
                // 提交的偏移量是下一條要讀取的訊息
                Map<TopicPartition, OffsetAndMetadata> offsets =
                    Collections.singletonMap(
                        new TopicPartition(record.topic(), record.partition()),
                        new OffsetAndMetadata(record.offset() + 1)
                    );

                // 提交該訊息的偏移量
                consumer.commitSync(offsets);
            }
        }
    }

    // 按批次提交偏移量
    // 每處理指定數量的訊息後提交一次
    public void batchCommit(KafkaConsumer<String, String> consumer) {
        int count = 0;
        final int commitInterval = 100;
        Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();

        while (true) {
            ConsumerRecords<String, String> records =
                consumer.poll(Duration.ofMillis(100));

            for (ConsumerRecord<String, String> record : records) {
                processRecord(record);

                // 更新當前偏移量
                currentOffsets.put(
                    new TopicPartition(record.topic(), record.partition()),
                    new OffsetAndMetadata(record.offset() + 1)
                );

                count++;

                // 每處理指定數量後提交
                if (count % commitInterval == 0) {
                    consumer.commitAsync(currentOffsets, null);
                }
            }
        }
    }

    private void processRecord(ConsumerRecord<String, String> record) {
        // 實現訊息處理邏輯
        System.out.println("處理訊息: " + record.value());
    }
}

分區分配策略決定了消費者群組中的分區如何分配給各個消費者。Kafka提供了多種內建的分配策略，每種策略有其特定的適用場景。Range策略按主題將分區連續分配給消費者，簡單但可能導致負載不均。RoundRobin策略以輪詢方式分配，實現更均衡的負載分佈。Sticky策略在保持均衡的同時盡量減少重新平衡時的分區移動，降低重新平衡的影響。CooperativeSticky策略支援增量式重新平衡，允許消費者在重新平衡期間繼續消費未受影響的分區。

// 分區分配策略配置範例
// 展示不同分配策略的配置和使用場景

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.CooperativeStickyAssignor;
import org.apache.kafka.clients.consumer.RangeAssignor;
import org.apache.kafka.clients.consumer.RoundRobinAssignor;
import org.apache.kafka.clients.consumer.StickyAssignor;

import java.util.Properties;

public class PartitionAssignmentConfig {

    // Range 分配策略
    // 按主題將分區連續分配給消費者
    // 適用於需要按主題處理的場景
    public Properties createRangeAssignmentConfig() {
        Properties props = new Properties();

        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "range-group");

        // 設定 Range 分配策略
        // 這是預設的分配策略
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
            RangeAssignor.class.getName());

        return props;
    }

    // RoundRobin 分配策略
    // 以輪詢方式將所有分區分配給消費者
    // 實現更均衡的負載分佈
    public Properties createRoundRobinAssignmentConfig() {
        Properties props = new Properties();

        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "roundrobin-group");

        // 設定 RoundRobin 分配策略
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
            RoundRobinAssignor.class.getName());

        return props;
    }

    // Sticky 分配策略
    // 在保持均衡的同時盡量減少分區移動
    // 減少重新平衡的影響
    public Properties createStickyAssignmentConfig() {
        Properties props = new Properties();

        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "sticky-group");

        // 設定 Sticky 分配策略
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
            StickyAssignor.class.getName());

        return props;
    }

    // CooperativeSticky 分配策略（推薦）
    // 支援增量式重新平衡
    // 消費者可以在重新平衡期間繼續消費未受影響的分區
    public Properties createCooperativeStickyAssignmentConfig() {
        Properties props = new Properties();

        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "cooperative-sticky-group");

        // 設定 CooperativeSticky 分配策略
        // 這是 Kafka 2.4+ 推薦的策略
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
            CooperativeStickyAssignor.class.getName());

        return props;
    }

    // 使用多個分配策略
    // 消費者群組會選擇所有成員都支援的策略
    public Properties createMultipleAssignmentStrategiesConfig() {
        Properties props = new Properties();

        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "multi-strategy-group");

        // 設定多個分配策略
        // 群組會選擇第一個所有成員都支援的策略
        String strategies = String.join(",",
            CooperativeStickyAssignor.class.getName(),
            StickyAssignor.class.getName(),
            RoundRobinAssignor.class.getName(),
            RangeAssignor.class.getName()
        );
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, strategies);

        return props;
    }
}

靜態成員資格是Kafka 2.3引入的功能，用於減少不必要的重新平衡。當消費者配置了group.instance.id時，它會成為群組的靜態成員。靜態成員在重啟時會重新獲得之前的分區分配，而不會觸發重新平衡，只要重啟時間不超過session.timeout.ms。這對於需要維護本地狀態或快取的應用程式特別有用，因為可以避免在重啟後重建狀態。

// 靜態成員配置範例
// 展示如何配置靜態群組成員以減少重新平衡

import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.util.Properties;
import java.util.UUID;

public class StaticMembershipConfig {

    // 建立靜態成員配置
    public Properties createStaticMemberConfig(String instanceId) {
        Properties props = new Properties();

        // 基本配置
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "static-members-group");

        // 設定靜態成員 ID
        // 這個 ID 必須在群組中唯一
        // 相同 ID 的消費者重啟後會獲得相同的分區分配
        props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, instanceId);

        // 設定較長的會話超時
        // 允許更長的重啟時間而不觸發重新平衡
        // 靜態成員在重啟期間會保持其分區分配
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "300000"); // 5分鐘

        // 設定心跳間隔
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "60000"); // 1分鐘

        return props;
    }

    // 建立基於主機名的靜態成員配置
    // 適用於部署在固定主機上的消費者
    public Properties createHostBasedStaticMemberConfig() {
        Properties props = new Properties();

        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "host-based-group");

        // 使用主機名作為實例 ID
        String hostname = getHostname();
        props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, hostname);

        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "300000");

        return props;
    }

    // 建立基於 Pod 名稱的靜態成員配置
    // 適用於 Kubernetes StatefulSet 部署
    public Properties createK8sStatefulSetConfig() {
        Properties props = new Properties();

        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "k8s-statefulset-group");

        // 在 Kubernetes StatefulSet 中，Pod 名稱是穩定的
        // 例如：myapp-0, myapp-1, myapp-2
        String podName = System.getenv("HOSTNAME");
        if (podName != null) {
            props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, podName);
        }

        // 設定會話超時長度要考慮 Pod 重啟時間
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "300000");

        return props;
    }

    private String getHostname() {
        try {
            return java.net.InetAddress.getLocalHost().getHostName();
        } catch (Exception e) {
            // 如果無法獲取主機名，生成一個唯一 ID
            return "consumer-" + UUID.randomUUID().toString().substring(0, 8);
        }
    }
}

跨區域消費是在分散式系統中常見的需求。當消費者和Broker位於不同的資料中心或可用區域時，可以透過配置client.rack來從最近的副本讀取資料，降低網路延遲和流量成本。這需要Broker端也正確配置了副本的rack資訊，並啟用了從最近副本擷取的功能。

效能監控和調校是確保Kafka消費者高效運作的重要環節。監控指標包括消費延遲、處理速率、重新平衡頻率以及提交延遲等。根據這些指標，可以調整擷取參數、心跳設定以及處理邏輯。常見的效能問題包括消費者滯後、頻繁重新平衡以及處理超時，這些問題通常可以透過調整配置參數來解決。

# Kafka 消費者監控工具
# 這個 Python 腳本展示了如何監控消費者的關鍵指標

from kafka import KafkaConsumer, KafkaAdminClient
from kafka.admin import ConsumerGroupDescription
import time
from typing import Dict, List, Any
from dataclasses import dataclass
import json

@dataclass
class ConsumerMetrics:
    """
    消費者指標資料類別

    儲存消費者的各項效能指標
    """
    # 群組 ID
    group_id: str
    # 成員數量
    member_count: int
    # 分區分配
    partition_assignment: Dict[str, List[int]]
    # 消費延遲
    total_lag: int
    # 每個主題的延遲
    lag_by_topic: Dict[str, int]

class KafkaConsumerMonitor:
    """
    Kafka 消費者監控器

    提供監控消費者群組狀態和效能的功能
    """

    def __init__(self, bootstrap_servers: str):
        """
        初始化監控器

        參數:
            bootstrap_servers: Kafka 叢集連接地址
        """
        self.bootstrap_servers = bootstrap_servers
        # 建立管理客戶端
        self.admin_client = KafkaAdminClient(
            bootstrap_servers=bootstrap_servers
        )

    def get_consumer_group_info(
        self,
        group_id: str
    ) -> Dict[str, Any]:
        """
        取得消費者群組資訊

        參數:
            group_id: 消費者群組 ID

        返回:
            包含群組資訊的字典
        """
        # 描述消費者群組
        groups = self.admin_client.describe_consumer_groups([group_id])

        if not groups:
            return {"error": "群組不存在"}

        group = groups[0]

        # 收集群組資訊
        info = {
            "group_id": group.group_id,
            "state": group.state,
            "protocol_type": group.protocol_type,
            "protocol": group.protocol,
            "coordinator": {
                "host": group.coordinator.host,
                "port": group.coordinator.port
            },
            "members": []
        }

        # 收集成員資訊
        for member in group.members:
            member_info = {
                "member_id": member.member_id,
                "client_id": member.client_id,
                "client_host": member.client_host,
                "group_instance_id": member.group_instance_id,
                "assignment": self._parse_assignment(member.member_assignment)
            }
            info["members"].append(member_info)

        return info

    def get_consumer_lag(
        self,
        group_id: str,
        topics: List[str] = None
    ) -> Dict[str, Any]:
        """
        計算消費者延遲

        參數:
            group_id: 消費者群組 ID
            topics: 要檢查的主題列表

        返回:
            包含延遲資訊的字典
        """
        # 取得群組的偏移量
        offsets = self.admin_client.list_consumer_group_offsets(group_id)

        # 取得主題的最新偏移量
        # 需要建立消費者來取得
        consumer = KafkaConsumer(
            bootstrap_servers=self.bootstrap_servers,
            group_id=None  # 不加入任何群組
        )

        lag_info = {
            "group_id": group_id,
            "total_lag": 0,
            "partitions": []
        }

        for tp, offset_meta in offsets.items():
            # 檢查是否在指定的主題中
            if topics and tp.topic not in topics:
                continue

            # 取得分區的最新偏移量
            end_offsets = consumer.end_offsets([tp])
            end_offset = end_offsets.get(tp, 0)

            # 計算延遲
            current_offset = offset_meta.offset
            lag = max(0, end_offset - current_offset)

            partition_info = {
                "topic": tp.topic,
                "partition": tp.partition,
                "current_offset": current_offset,
                "end_offset": end_offset,
                "lag": lag
            }
            lag_info["partitions"].append(partition_info)
            lag_info["total_lag"] += lag

        consumer.close()

        return lag_info

    def monitor_consumer_groups(
        self,
        group_ids: List[str],
        interval_seconds: int = 30
    ):
        """
        持續監控消費者群組

        參數:
            group_ids: 要監控的群組 ID 列表
            interval_seconds: 監控間隔（秒）
        """
        print("開始監控消費者群組...")
        print(f"監控間隔: {interval_seconds} 秒")
        print("-" * 60)

        while True:
            for group_id in group_ids:
                try:
                    # 取得群組資訊
                    info = self.get_consumer_group_info(group_id)

                    # 取得延遲資訊
                    lag = self.get_consumer_lag(group_id)

                    # 輸出監控結果
                    print(f"\n時間: {time.strftime('%Y-%m-%d %H:%M:%S')}")
                    print(f"群組: {group_id}")
                    print(f"狀態: {info.get('state', 'Unknown')}")
                    print(f"成員數: {len(info.get('members', []))}")
                    print(f"總延遲: {lag.get('total_lag', 0)}")

                    # 檢查是否有高延遲的分區
                    for partition in lag.get("partitions", []):
                        if partition["lag"] > 1000:
                            print(
                                f"  警告: {partition['topic']}-"
                                f"{partition['partition']} "
                                f"延遲 {partition['lag']}"
                            )

                except Exception as e:
                    print(f"監控 {group_id} 時發生錯誤: {str(e)}")

            print("-" * 60)
            time.sleep(interval_seconds)

    def _parse_assignment(self, assignment: bytes) -> Dict[str, List[int]]:
        """
        解析成員分配資訊

        參數:
            assignment: 原始分配位元組

        返回:
            主題到分區列表的對映
        """
        # 簡化處理，實際需要解析 Kafka 協議格式
        return {}

    def get_group_metrics(self, group_id: str) -> ConsumerMetrics:
        """
        取得消費者群組的綜合指標

        參數:
            group_id: 消費者群組 ID

        返回:
            ConsumerMetrics 物件
        """
        info = self.get_consumer_group_info(group_id)
        lag = self.get_consumer_lag(group_id)

        # 計算每個主題的延遲
        lag_by_topic = {}
        for partition in lag.get("partitions", []):
            topic = partition["topic"]
            if topic not in lag_by_topic:
                lag_by_topic[topic] = 0
            lag_by_topic[topic] += partition["lag"]

        # 建構分區分配對映
        partition_assignment = {}
        for member in info.get("members", []):
            member_id = member["client_id"]
            partition_assignment[member_id] = member.get("assignment", {})

        return ConsumerMetrics(
            group_id=group_id,
            member_count=len(info.get("members", [])),
            partition_assignment=partition_assignment,
            total_lag=lag.get("total_lag", 0),
            lag_by_topic=lag_by_topic
        )

# 使用範例
if __name__ == "__main__":
    # 建立監控器
    monitor = KafkaConsumerMonitor("localhost:9092")

    # 取得群組資訊
    info = monitor.get_consumer_group_info("my-consumer-group")
    print("群組資訊:")
    print(json.dumps(info, indent=2, ensure_ascii=False))

    # 取得延遲資訊
    lag = monitor.get_consumer_lag("my-consumer-group")
    print("\n延遲資訊:")
    print(json.dumps(lag, indent=2, ensure_ascii=False))

    # 持續監控
    # monitor.monitor_consumer_groups(
    #     ["my-consumer-group", "another-group"],
    #     interval_seconds=30
    # )

錯誤處理和重試機制是建構可靠消費者的關鍵。消費者可能遇到的錯誤包括反序列化失敗、處理異常以及偏移量提交失敗等。對於可重試的錯誤，應該實現適當的重試邏輯。對於不可恢復的錯誤，應該將問題訊息發送到死信佇列以便後續處理。良好的錯誤處理策略可以確保系統在遇到問題時能夠優雅地降級，而不是完全失敗。

消費者的優雅關閉是另一個重要的考量。當消費者需要關閉時，應該先停止輪詢新訊息，完成當前正在處理的訊息，提交偏移量，然後關閉消費者。使用wakeup方法可以安全地中斷poll調用，使消費者能夠快速響應關閉請求。正確的關閉流程可以防止訊息遺失和重複消費。

Kafka消費者的配置和使用涉及眾多相互關聯的參數和概念。從基本的連接配置到進階的分區分配策略，每個配置都會影響消費者的行為和效能。正確理解這些配置的含義和相互作用，根據應用程式的具體需求進行調整，是建構高效可靠的Kafka消費者應用程式的基礎。隨著訊息量的增長和業務需求的變化，持續監控和調校消費者配置是維持系統健康運作的重要工作。

總結而言，Kafka消費者是串流處理架構中的核心元件，其正確配置和使用對系統的效能和可靠性有著重要影響。本文深入探討了消費者的運作機制、關鍵配置參數、偏移量管理策略以及分區分配策略，提供了豐富的程式碼範例和最佳實踐建議。讀者應該根據自己的應用場景和需求，選擇合適的配置和策略，並透過持續監控來最佳化消費者的效能。掌握這些知識將幫助開發者建構穩定、高效且可擴展的串流處理應用程式。

玄貓 BlackCat

玄貓 BlackCat

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。