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
