Sarama Kafka Consumer Group函数的输出

在 Go 语言中使用 Sarama 库处理来自 Kafka 的消息时，需要将原始代码中的 partitionconsumer 替换为 sarama newconsumergroup 包。为此，需要创建一个 consumergroup，并在其 ConsumeClaim 方法中处理消息。新创建的 KafkaInput 结构应包含一个消息通道，并通过 goroutine 从消费者组接收消息并将其推送到该通道。最终，Read 方法将从该通道中读取消息并将其转换为 HTTP 有效负载。

我对 go lang 非常陌生，并尝试对使用 sarama 库使用来自 kafka 的消息的开源库进行一些调整。原始代码可以在这里找到。

原始包实现了一个 partitionconsumer，如果不需要在使用同一主题的多个消费者之间保持读取一致性，它就可以正常工作，但是，这对我来说不起作用。

我在同一个应用程序中完成了一些工作，使用我在网上找到的一些示例来实现 sarama newconsumergroup 包。

下面是我当前正在运行的代码：

package main

import (
    "context"
    // "flag"
    "os"
    "os/signal"
    "sync"
    "syscall"

    "encoding/json"
    "log"
    "strings"

    "github.com/shopify/sarama"
    // "github.com/shopify/sarama/mocks"
)

// kafkainput is used for recieving kafka messages and
// transforming them into http payloads.
type kafkainput struct {
    config    *kafkaconfig
    // consumers []sarama.partitionconsumer
    messages  chan *sarama.consumermessage
}

var (
    brokers  = ""
    version  = ""
    group    = ""
    topics   = ""
    assignor = ""
    oldest   = true
    verbose  = false
)

// consumer represents a sarama consumer group consumer
type consumer struct {
    ready chan bool
}

// newkafkainput creates instance of kafka consumer client.
func newkafkainput(address string, config *kafkaconfig) *kafkainput {
    /**
     * construct a new sarama configuration.
     * the kafka cluster version has to be defined before the consumer/producer is initialized.
     */
    c := sarama.newconfig()
    // configuration options go here

    log.println("starting a new sarama consumer")

    if verbose {
        sarama.logger = log.new(os.stdout, "[sarama] ", log.lstdflags)
    }

    version, err := sarama.parsekafkaversion("2.1.1")
    if err != nil {
        log.panicf("error parsing kafka version: %v", err)
    }

    c.version = version

    if oldest {
        c.consumer.offsets.initial = sarama.offsetoldest
    }

    /**
     * setup a new sarama consumer group
     */
    consumer := consumer{ready: make(chan bool)}

    ctx, cancel := context.withcancel(context.background())
    client, err := sarama.newconsumergroup(strings.split(config.host, ","), config.group, c)

    if err != nil {
        log.panicf("error creating consumer group client: %v", err)
    }

    wg := &sync.waitgroup{}
    wg.add(1)
    go func() {
        defer wg.done()
        for {
            if err := client.consume(ctx, []string{config.topic}, &consumer); err != nil {
                log.panicf("error from consumer: %v", err)
            }

            // check if context was cancelled, signaling that the consumer should stop
            if ctx.err() != nil {
                return
            }

            consumer.ready = make(chan bool)
        }

    }()

    <-consumer.ready // await till the consumer has been set up
    log.println("sarama consumer up and running!...")

    sigterm := make(chan os.signal, 1)
    signal.notify(sigterm, syscall.sigint, syscall.sigterm)
    select {
    case <-ctx.done():
        log.println("terminating: context cancelled")
    case <-sigterm:
        log.println("terminating: via signal")
    }
    cancel()
    wg.wait()
    if err = client.close(); err != nil {
        log.panicf("error closing client: %v", err)
    }

    i := &kafkainput{
        config: config,
        // consumers: make([]sarama.partitionconsumer, len(partitions)),
        // messages:  make(chan *sarama.consumermessage, 256),
        messages: make(chan *sarama.consumermessage, 256),
    }


    return i
}

// consumeclaim must start a consumer loop of consumergroupclaim's messages().
func (consumer *consumer) consumeclaim(session sarama.consumergroupsession, claim sarama.consumergroupclaim) error {
    // note:
    // do not move the code below to a goroutine.
    // the `consumeclaim` itself is called within a goroutine, see:
    // https://github.com/shopify/sarama/blob/master/consumer_group.go#l27-l29
    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
}

// errorhandler should receive errors
func (i *kafkainput) errorhandler(consumer sarama.partitionconsumer) {
    for err := range consumer.errors() {
        log.println("failed to read access log entry:", err)
    }
}

// read comment

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

}

func (i *kafkainput) string() string {
    return "kafka input: " + i.config.host + "/" + i.config.topic
}

// setup is run at the beginning of a new session, before consumeclaim
func (consumer *consumer) setup(sarama.consumergroupsession) error {
    // mark the consumer as ready
    close(consumer.ready)
    return nil
}

// cleanup is run at the end of a session, once all consumeclaim goroutines have exited
func (consumer *consumer) cleanup(sarama.consumergroupsession) error {
    return nil
}

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
}

... 
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

}

我也完全有可能把这件事弄得一团糟，但是任何帮助和意见都会受到赞赏。

谢谢

解决方案

这是我的问题的解决方案。我的 goroutine 阻塞了主函数，需要将它们分解。如果下面的代码没有任何意义，这里是我正在修改的程序的链接：https://github.com/buger/goreplay。如果我能得到所有者的回复，我计划清理代码并提交拉取请求，或者可能发布叉。

package main

import (
    "context"
    "encoding/json"
    "strings"

    "os"

    "log"

    "github.com/Shopify/sarama"
)

// KafkaInput is used for recieving Kafka messages and
// transforming them into HTTP payloads.
type KafkaInput struct {
    sarama.ConsumerGroup
    config   *KafkaConfig
    consumer Consumer
    messages chan *sarama.ConsumerMessage
}

// Consumer represents a Sarama consumer group consumer
type Consumer struct {
    ready    chan bool
    messages chan *sarama.ConsumerMessage
}

var (
    brokers  = ""
    version  = ""
    group    = ""
    topics   = ""
    assignor = ""
    oldest   = true
    verbose  = false
)

// NewKafkaInput creates instance of kafka consumer client.
func NewKafkaInput(address string, config *KafkaConfig) *KafkaInput {
    /**
     * Construct a new Sarama configuration.
     * The Kafka cluster version has to be defined before the consumer/producer is initialized.
     */
    c := sarama.NewConfig()
    // Configuration options go here

    log.Printf("KafkaConfig: %s", config.host)
    log.Printf("KafkaConfig: %s", config.group)
    log.Printf("KafkaConfig: %s", config.topic)

    log.Println("Starting a new Sarama consumer")

    if verbose {
        sarama.Logger = log.New(os.Stdout, "[sarama] ", log.LstdFlags)
    }

    version, err := sarama.ParseKafkaVersion("2.1.1")
    if err != nil {
        log.Panicf("Error parsing Kafka version: %v", err)
    }

    c.Version = version

    if oldest {
        c.Consumer.Offsets.Initial = sarama.OffsetOldest
    }

    group, err := sarama.NewConsumerGroup(strings.Split(config.host, ","), config.group, c)

    /**
     * Setup a new Sarama consumer group
     */
    consumer := Consumer{
        ready:    make(chan bool),
        messages: make(chan *sarama.ConsumerMessage, 256),
    }

    i := &KafkaInput{
        ConsumerGroup: group,
        config:        config,
        messages:      make(chan *sarama.ConsumerMessage, 256),
        consumer:      consumer,
    }

    go i.loop([]string{config.topic})
    i.messages = consumer.messages
    return i
}

//ConsumeClaim and stuff
func (i *KafkaInput) ConsumeClaim(s sarama.ConsumerGroupSession, c sarama.ConsumerGroupClaim) error {
    for msg := range c.Messages() {
        s.MarkMessage(msg, "")
        i.Push(msg)
    }
    return nil
}

func (i *KafkaInput) loop(topic []string) {
    ctx := context.Background()
    for {
        if err := i.Consume(ctx, []string{i.config.topic}, i); err != nil {
            return
        }
    }
}

// Push Messages
func (i *KafkaInput) Push(m *sarama.ConsumerMessage) {
    if i.consumer.messages != nil {
        log.Printf("MSGPUSH: %s", m)
        i.consumer.messages <- m
    }
}

func (i *KafkaInput) Read(data []byte) (int, error) {

    message := <-i.messages
    log.Printf("Msg: %s", string(message.Value))
    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

}

func (i *KafkaInput) String() string {
    return "Kafka Input: " + i.config.host + "/" + i.config.topic
}

// Setup is run at the beginning of a new session, before ConsumeClaim
func (i *KafkaInput) Setup(s sarama.ConsumerGroupSession) error {
    return nil
}

// Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited
func (i *KafkaInput) Cleanup(s sarama.ConsumerGroupSession) error {
    return nil
}

今天关于《Sarama Kafka Consumer Group函数的输出》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！