匯合 Kafka Go 客戶端中的'無效複製因子”

php小編西瓜今天為大家帶來一篇關於Kafka Go客戶端中的「無效複製因子」的文章。 Kafka是一個高效能、可擴展的分散式串流處理平台，而Go則是一種簡潔、有效率的程式語言。本文將重點放在Kafka Go客戶端中出現的「無效複製因子」問題，探索其原因和解決方法，幫助讀者更好地理解和應對這一常見的技術挑戰。無效複製因子可能導致資料不一致和效能下降，因此了解如何處理此問題對於Kafka使用者來說非常重要。接下來就讓我們一起來深入探索吧！

問題內容

我是卡夫卡新手，正在嘗試開始我的專案。我的 docker-compose.yml 中有這個

version: '3'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  broker:
    image: confluentinc/cp-kafka:7.3.0
    container_name: broker
    depends_on:
      - zookeeper
    ports:
      - 9092:9092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1

然後我運行我的 main.go 文件，其中包含生產者和消費者以及一些模擬主題。

package main

import (
    "fmt"
    "log"
    "time"

    "github.com/confluentinc/confluent-kafka-go/kafka"
)



func main()  {
    topic := "HVSE"
    p, err := kafka.NewProducer(&kafka.ConfigMap{
        "bootstrap.servers": "localhost:9092",
        "client.id": "foo",
        "acks": "all",
    })
    go func() {
        consumer, err := kafka.NewConsumer(&kafka.ConfigMap{
            "bootstrap.servers":    "localhost:9092",
            "group.id":             "foo",
            "auto.offset.reset":    "smallest",
        })

        if err != nil {
            log.Fatal(err)
        }

        err = consumer.Subscribe(topic, nil)

        if err != nil {
            log.Fatal(err)
        }

        for {
                ev := consumer.Poll(100)
                // fmt.Println(ev)
                switch e := ev.(type) {
                case *kafka.Message:
                        fmt.Printf("consumed message from queue: %s\n", string(e.Value))
                case *kafka.Error:
                        fmt.Printf("%v\n", e)
                        // return
                // default:
                //      fmt.Printf("Ignored %v\n", e)
                }
        }
    }()

    deliverch := make(chan kafka.Event, 10000)
    for {
        err = p.Produce(&kafka.Message{
            TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
            Value: []byte("FOO"),
        },
            deliverch,
        )
        if err != nil {
            log.Fatal(err)
        }
    
        <- deliverch
        time.Sleep(time.Second * 1)
    }
}

如果我取消預設值的註釋，我就會進入它。

否則我會在控制台中收到此錯誤。

2023/09/26 13:45:05 Broker: Invalid replication factor
exit status 1

我的 kafka 和 Zookeeper 容器正在運作。

我更改了 docker-compose.yml 文件，但這沒有幫助。我發現我的consumer.Events()為零，但我不明白為什麼會這樣

解決方法

我複製了您的程式碼，它是正確的，但compose.yml 中的問題僅調整了ADVERTISED_LISTENERS

而不是這個：

KAFKA_ADVERTISED_LISTENERS:PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092

使用這個：

KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092

以上是匯合 Kafka Go 客戶端中的'無效複製因子”的詳細內容。更多資訊請關注PHP中文網其他相關文章！