如何从头订阅

Question

我正在尝试使用 GroupId 编写一个 Kafka Consumer foo，它订阅某个主题并从头开始读取（即使之前有偏移量）。我尝试与重新平衡回调一起使用Subscribe，但它似乎从未被调用（已设置设置go.application）。有什么例子可以使这项工作成功吗？

Answer 1

你可能只需要将你的值设置 auto.offset.reset为kafka.OffsetBeginning.String()：

package main

/**

 * Copyright 2016 Confluent Inc.

 */

// consumer_example implements a consumer using the non-channel Poll() API

// to retrieve messages and events.

import (

    "fmt"

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

    "os"

    "os/signal"

    "syscall"

)

func main() {

    broker := "YOUR_BROKER"

    group := "YOUR_GROUP"

    topics := "YOUR_TOPICS"

    sigchan := make(chan os.Signal, 1)

    signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)

    c, err := kafka.NewConsumer(&kafka.ConfigMap{

        "bootstrap.servers":  broker,

        "group.id":           group,

        "session.timeout.ms": 6000,

        "auto.offset.reset":  kafka.OffsetBeginning.String()})

    if err != nil {

        fmt.Fprintf(os.Stderr, "Failed to create consumer: %s\n", err)

        os.Exit(1)

    }

    fmt.Printf("Created Consumer %v\n", c)

    err = c.SubscribeTopics(topics, nil)

    run := true

    for run == true {

        select {

        case sig := <-sigchan:

            fmt.Printf("Caught signal %v: terminating\n", sig)

            run = false

        default:

            ev := c.Poll(100)

            if ev == nil {

                continue

            }

            switch e := ev.(type) {

            case *kafka.Message:

                fmt.Printf("%% Message on %s:\n%s\n",

                    e.TopicPartition, string(e.Value))

                if e.Headers != nil {

                    fmt.Printf("%% Headers: %v\n", e.Headers)

                }

            case kafka.Error:

                // Errors should generally be considered as informational, the client will try to automatically recover

                fmt.Fprintf(os.Stderr, "%% Error: %v\n", e)

            default:

                fmt.Printf("Ignored %v\n", e)

            }

        }

    }

    fmt.Printf("Closing consumer\n")

    c.Close()

}

Answer 2

我们现在设置enable.auto.commit为false. 这样，就不会存储偏移量，我们每次运行都从头开始消费。