快捷導(dǎo)航

Go語言使用kafka-go實現(xiàn)Kafka消費消息

更新時間：2024年12月16日 11:28:14 作者：宋發(fā)元

本篇文章主要介紹了使用kafka-go庫消費Kafka消息,包含F(xiàn)etchMessage和ReadMessage的區(qū)別和適用場景,文中通過示例代碼介紹的非常詳細(xì),具有一定的參考價值,感興趣的可以了解一下

安裝 kafka-go 庫

首先，你需要在項目中安裝 kafka-go 庫。可以使用以下命令：

go get github.com/segmentio/kafka-go

初始化 Kafka Reader

為了從 Kafka 消費消息，我們首先需要配置和初始化 Kafka Reader。以下是一個簡單的 Kafka Reader 初始化示例：

package main

import (
    "context"
    "log"
    "github.com/segmentio/kafka-go"
)

func main() {
    // 創(chuàng)建 Kafka Reader
    kafkaReader := kafka.NewReader(kafka.ReaderConfig{
        Brokers:   []string{"localhost:9092"}, // Kafka broker 地址
        Topic:     "example-topic",            // 訂閱的 Kafka topic
        GroupID:   "example-group",            // 消費者組 ID
        Partition: 0,                          // 分區(qū)號 (可選)
        MinBytes:  10e3,                       // 10KB
        MaxBytes:  10e6,                       // 10MB
    })

    defer kafkaReader.Close()
}

使用 FetchMessage 消費消息

FetchMessage 允許你從 Kafka 消費消息并手動提交偏移量，這給你對消息處理的更精確控制。以下是如何使用 FetchMessage 的示例：

func consumeWithFetchMessage() {
    ctx := context.Background()
    
    for {
        // 從 Kafka 中獲取下一條消息
        m, err := kafkaReader.FetchMessage(ctx)
        if err != nil {
            log.Printf("獲取消息時出錯: %v", err)
            break
        }

        // 打印消息內(nèi)容
        log.Printf("消息: %s, 偏移量: %d", string(m.Value), m.Offset)

        // 處理消息 (在這里可以進(jìn)行你的業(yè)務(wù)邏輯)

        // 手動提交偏移量
        if err := kafkaReader.CommitMessages(ctx, m); err != nil {
            log.Printf("提交偏移量時出錯: %v", err)
        }
    }
}

優(yōu)點

精確控制偏移量：在處理消息后，你可以手動選擇是否提交偏移量，這樣可以確保只有在消息處理成功后才提交。
重試機制：可以靈活地處理失敗消息，例如在處理失敗時，不提交偏移量，從而實現(xiàn)消息的重新消費。

缺點

代碼復(fù)雜度增加：需要手動處理偏移量提交，會增加一些額外的代碼量。

使用 ReadMessage 消費消息

ReadMessage 是一種更簡單的方式，從 Kafka 中獲取消息并自動提交偏移量。適用于對消費邏輯不太敏感的場景。以下是使用 ReadMessage 的示例：

func consumeWithReadMessage() {
    ctx := context.Background()
    
    for {
        // 從 Kafka 中讀取下一條消息并自動提交偏移量
        dataInfo, err := kafkaReader.ReadMessage(ctx)
        if err != nil {
            log.Printf("讀取消息時出錯: %v", err)
            break
        }

        // 打印消息內(nèi)容
        log.Printf("消息: %s, 偏移量: %d", string(dataInfo.Value), dataInfo.Offset)

        // 處理消息 (在這里可以進(jìn)行你的業(yè)務(wù)邏輯)
    }
}

優(yōu)點

簡單易用：ReadMessage 自動提交偏移量，代碼簡潔，易于維護。
快速開發(fā)：適合簡單的消息處理邏輯和對消息可靠性要求不高的場景。

缺點

缺乏靈活性：無法在處理失敗時重新消費消息，因為偏移量已經(jīng)自動提交。

總結(jié)選擇

方法	優(yōu)點	缺點	適用場景
`FetchMessage`	需要手動提交偏移量，精確控制消息處理和提交邏輯	代碼復(fù)雜度較高	需要精確控制消息處理的場景，例如處理失敗重試
`ReadMessage`	簡單易用，自動提交偏移量，代碼更簡潔	無法重新消費已處理失敗的消息	簡單的消息處理，對消息處理成功率要求不高的場景

完整示例

以下是一個完整的 Kafka 消費者示例，包括 FetchMessage 和 ReadMessage 兩種方法。可以根據(jù)你的需求選擇合適的方法：

package main

import (
    "context"
    "log"
    "github.com/segmentio/kafka-go"
)

func main() {
    // 創(chuàng)建 Kafka Reader
    kafkaReader := kafka.NewReader(kafka.ReaderConfig{
        Brokers:   []string{"localhost:9092"},
        Topic:     "example-topic",
        GroupID:   "example-group",
        MinBytes:  10e3, // 10KB
        MaxBytes:  10e6, // 10MB
    })

    defer kafkaReader.Close()

    // 使用 FetchMessage 消費消息
    log.Println("開始使用 FetchMessage 消費 Kafka 消息...")
    consumeWithFetchMessage(kafkaReader)

    // 使用 ReadMessage 消費消息
    log.Println("開始使用 ReadMessage 消費 Kafka 消息...")
    consumeWithReadMessage(kafkaReader)
}

func consumeWithFetchMessage(kafkaReader *kafka.Reader) {
    ctx := context.Background()

    for {
        m, err := kafkaReader.FetchMessage(ctx)
        if err != nil {
            log.Printf("FetchMessage 獲取消息時出錯: %v", err)
            break
        }

        log.Printf("FetchMessage 消息: %s, 偏移量: %d", string(m.Value), m.Offset)

        // 手動提交偏移量
        if err := kafkaReader.CommitMessages(ctx, m); err != nil {
            log.Printf("FetchMessage 提交偏移量時出錯: %v", err)
        }
    }
}

func consumeWithReadMessage(kafkaReader *kafka.Reader) {
    ctx := context.Background()

    for {
        dataInfo, err := kafkaReader.ReadMessage(ctx)
        if err != nil {
            log.Printf("ReadMessage 讀取消息時出錯: %v", err)
            break
        }

        log.Printf("ReadMessage 消息: %s, 偏移量: %d", string(dataInfo.Value), dataInfo.Offset)
    }
}