我對 Go Lang 非常陌生，并試圖對使用 Sarama 庫從 Kafka 消費消息的開源庫進行一些調整。原始代碼可以在這里找到。原始包實現了一個 PartitionConsumer，如果不需要跨多個消費者使用同一主題的讀取一致性，它就可以正常工作，但是，這對我不起作用。我在同一個應用程序中做了一些工作，使用我在網上找到的一些示例來實現 sarama NewConsumerGroup 包。KafkaConfig 為消費者攜帶 groupID 和 Topic。當我運行這個程序時，消費者啟動并使用正確的組從正確的主題中讀取，并使用在此函數中創建的 ConsumerClaim 將其打印到 STDOUT：func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {    for message := range claim.Messages() {        log.Printf("Message claimed: value = %s, timestamp = %v, topic = %s", string(message.Value), message.Timestamp, message.Topic)        session.MarkMessage(message, "")    }    return nil}然而，我認為我需要的是NewKafkaInput函數返回*KafkaInput添加到結構的聲明中的消息（如果我在這里使用了錯誤的術語，請原諒我，這是我的第一個 Go 牛仔競技表演）。... i := &KafkaInput{        config: config,        // consumers: make([]sarama.PartitionConsumer, len(partitions)),        // messages:  make(chan *sarama.ConsumerMessage, 256),        messages: make(chan *sarama.ConsumerMessage, 256),    }    return i}在此處完成的原始示例中：func NewKafkaInput(address string, config *KafkaConfig) *KafkaInput {    ...    go func(consumer sarama.PartitionConsumer) {                defer consumer.Close()                for message := range consumer.Messages() {                    i.messages <- message                }            }(consumer)    ...}我花了好幾天的時間來研究將函數移入和移出NewKafakInput函數，嘗試將消息添加到KafakInput函數外部的結構以及介于兩者之間的所有內容。我只是無法讓它工作。該NewKafakInput函數需要返回*KafkaInput任何消息，以便此函數可以完成：func (i *KafkaInput) Read(data []byte) (int, error) {    message := <-i.messages    if !i.config.useJSON {        copy(data, message.Value)        return len(message.Value), nil    }    var kafkaMessage KafkaMessage    json.Unmarshal(message.Value, &kafkaMessage)    buf, err := kafkaMessage.Dump()    if err != nil {        log.Println("Failed to decode access log entry:", err)        return 0, err    }    copy(data, buf)    return len(buf), nil}完全有可能我也把這件事弄得一團糟，但感謝任何幫助和輸入。