1

我在这里面临一个非常奇怪的问题。我在 go 中使用 sarama-cluster 库启动了一个 Kafka 消费者，以使用来自 kafka 主题的一些消息。但是正在启动的消费者没有收到消息。

然而，一件非常奇怪的事情正在发生。如果我启动另一个与之并行的消费者，则消息会突然传递给两个消费者。

我想不出一个合乎逻辑的解释。任何指针将不胜感激。

注意：这个问题是在 Kafka 和 Zookeeper 服务器非正常启动之后开始的。

下面是消费者的 go 代码，用于消费不工作的消息：

if err := consumer.Start(); err != nil {
    return err
}
updChan, err := consumer.Consume()
if err != nil {
    return err
}
go func() {
    for {
        select {
        case msg, ok := <-updChan:
            if !ok {
                return
            }
            var message liveupdater.KafkaMessage
            err := json.Unmarshal(msg.Msg, &message)
            if err != nil {
                fmt.Println(err)
            }
            err = handleMessaege(message)
            if err != nil {
                logrus.Println("encountered error:" + err.Error())
            }

            consumer.MarkProcessed(msg, string(message.Type))
        }
    }
}()

以下是消费者接收消息的 go 代码（此代码与之前代码之间的唯一区别是另一个消费者为同一主题并行运行）。

consumeMessages(config)

if err := consumer.Start(); err != nil {
    return err
}
updChan, err := consumer.Consume()
if err != nil {
    return err
}
go func() {
    for {
        select {
        case msg, ok := <-updChan:
            if !ok {
                return
            }
            var message liveupdater.KafkaMessage
            err := json.Unmarshal(msg.Msg, &message)
            if err != nil {
                fmt.Println(err)
            }
            err = handleMessaege(message)
            if err != nil {
                logrus.Println("encountered error:" + err.Error())
            }

            consumer.MarkProcessed(msg, string(message.Type))
        }
    }
}()

func consumeMessages(config *rakshak_config.Config) {
    kafkaConfig := kafka.Config{Brokers: strings.Split(config.Kafka.Brokers, ",")}
    logrus.Println("brokers %s", config.Kafka.Brokers)
    hermesConsumer, err := hermes.NewConsumer(hermes.Kafka, []string{config.Kafka.Topic}, kafkaConfig)
    if err != nil {
        logrus.Println("could not get consumer through hermes %s", err)
    }
    err = hermesConsumer.Start()
    if err != nil {
        logrus.Println("could not start consumer through hermes %s", err)
    }
    conChan, err := hermesConsumer.Consume()
    if err != nil {
        logrus.Println("not able to start consumer channel %s", err)
    }
    go func() {
        for {
            select {
            case msg, ok := <-conChan:
                if !ok {
                    logrus.Println("could not consume message")
                }

                logrus.Println("kafka msg string: %s", string(msg.Msg[:]))

                hermesConsumer.MarkProcessed(msg, "")
            }
        }
    }()

提前致谢。

4

0 回答 0