Apache Kafka 是一個強大的分布式流平臺，用于構建實時數(shù)據(jù)管道和流應用程序。在這篇博文中，我們將逐步使用 Golang 設置 Kafka 生產者和消費者。

先決條件

在我們開始之前，請確保您的計算機上安裝了以下軟件：

• Go（1.16 或更高）

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

• 卡夫卡

使用 Docker 設置 Kafka

為了快速設置 Kafka，我們將使用 Docker。在項目目錄中創(chuàng)建 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 中創(chuàng)建 Kafka 生產者

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

go mod init kafka-example

安裝 kafka-go 庫：

go get github.com/segmentio/kafka-go

現(xiàn)在，創(chuàng)建一個文件 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 主題發(fā)送 10 條消息。

運行生產者：

go run producer.go

您應該看到指示消息已生成的輸出。

在 Go 中創(chuàng)建 Kafka 消費者

創(chuàng)建文件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，您可以構建強大的、可擴展的實時數(shù)據(jù)處理系統(tǒng)。

隨意探索更高級的功能，例如消息分區(qū)、基于密鑰的消息分發(fā)以及與其他系統(tǒng)的集成。快樂編碼！

---

就是這樣！這篇博文簡要介紹了如何將 Kafka 與 Go 結合使用，非常適合想要開始實時數(shù)據(jù)處理的開發(fā)人員。

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