如何从头开始订阅

学习Golang要努力，但是不要急！今天的这篇文章《如何从头开始订阅》将会介绍到等等知识点，如果你想深入学习Golang，可以关注我！我会持续更新相关文章的，希望对大家都能有所帮助！

我正在尝试使用 GroupId foo 编写一个 Kafka Consumer，它订阅某个主题并从头开始读取（即使存在先前的偏移量） 。我尝试将 Subscribe 与重新平衡回调一起使用，但它似乎从未被调用（已设置 go.application 设置）。

有什么例子可以让这项工作发挥作用吗？

编辑： 添加了更多详细信息

解决方案

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

这个示例来自 Confluent Kafka go Github，您可能只需将 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()
}

好了，本文到此结束，带大家了解了《如何从头开始订阅》，希望本文对你有所帮助！关注golang学习网公众号，给大家分享更多Golang知识！