用 Go 建構 Kafka 生產者和消費者

Apache Kafka 是一個強大的分散式串流平台，用於建立即時資料管道和串流應用程式。在這篇文章中，我們將逐步使用 Golang 設定 Kafka 生產者和消費者。

先決條件

在我們開始之前，請確保您的電腦上安裝了以下軟體：

• Go（1.16 或更高）

• Docker（用於在本地運行 Kafka）

• 卡夫卡

使用 Docker 設定 Kafka

為了快速設定 Kafka，我們將使用 Docker。在專案目錄中建立 docker-compose.yml 檔案：

yamlCopy codeversion: '3.7'
services:
  zookeeper:
    image: wurstmeister/zookeeper:3.4.6
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka:2.13-2.7.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    depends_on:
      - zookeeper

執行以下指令啟動 Kafka 和 Zookeeper：

docker-compose up -d

在 Go 中建立 Kafka 生產者

首先，初始化一個新的 Go 模組：

go mod init kafka-example

安裝 kafka-go 函式庫：

go get github.com/segmentio/kafka-go

現在，建立一個檔案 Producer.go 並加入以下程式碼：

package main

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

func main() {
    writer := kafka.Writer{
        Addr:     kafka.TCP("localhost:9092"),
        Topic:    "example-topic",
        Balancer: &kafka.LeastBytes{},
    }

    defer writer.Close()

    for i := 0; i < 10; i++ {
        msg := kafka.Message{
            Key:   []byte(fmt.Sprintf("Key-%d", i)),
            Value: []byte(fmt.Sprintf("Hello Kafka %d", i)),
        }

        err := writer.WriteMessages(context.Background(), msg)
        if err != nil {
            log.Fatal("could not write message " + err.Error())
        }

        time.Sleep(1 * time.Second)
        fmt.Printf("Produced message: %s\n", msg.Value)
    }
}

此程式碼設定一個 Kafka 生產者，向 example-topic 主題發送 10 則訊息。

運作生產者：

go run producer.go

您應該看到指示訊息已產生的輸出。

在 Go 中創建 Kafka 消費者

建立檔案consumer.go並加入以下程式碼：

package main

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

func main() {
    reader := kafka.NewReader(kafka.ReaderConfig{
        Brokers: []string{"localhost:9092"},
        Topic:   "example-topic",
        GroupID: "example-group",
    })

    defer reader.Close()

    for {
        msg, err := reader.ReadMessage(context.Background())
        if err != nil {
            log.Fatal("could not read message " + err.Error())
        }
        fmt.Printf("Consumed message: %s\n", msg.Value)
    }
}

該消費者從 example-topic 主題讀取訊息並將其列印到控制台。

運行消費者：

go run consumer.go

您應該看到指示訊息已被消耗的輸出。

結論

在這篇文章中，我們示範如何使用 Golang 設定 Kafka 生產者和消費者。這個簡單的範例展示了產生和消費訊息的基礎知識，但 Kafka 的功能遠遠不止於此。借助 Kafka，您可以建立強大的、可擴展的即時數據處理系統。

隨意探索更高級的功能，例如訊息分區、基於金鑰的訊息分發以及與其他系統的整合。快樂編碼！

---

就是這樣！這篇部落格文章簡要介紹如何將 Kafka 與 Go 結合使用，非常適合想要開始即時資料處理的開發人員。

以上是用 Go 建構 Kafka 生產者和消費者的詳細內容。更多資訊請關注PHP中文網其他相關文章！