go - 如何从头订阅
问题描述
我正在尝试使用 GroupId 编写一个 Kafka Consumer foo
,它订阅某个主题并从一开始就读取(即使有以前的偏移量)。我尝试Subscribe
与重新平衡回调一起使用,但它似乎从未被调用(已设置go.application
设置)。
有什么例子可以使这项工作吗?
编辑:添加了更多细节
解决方案
这个例子来自Confluent Kafka go Github,你可能只需要设置 to 的 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()
}
推荐阅读
- node.js - 使用 nginx 反向代理在一个域上隐藏节点应用程序端口号
- ios - iOS React Native Firebase (Invertase) Google 登录构建失败
- python - 如何从 python Flask WTF 中的 SelectField 获取用户选择的值?
- dart - 将 ListTile 图标左对齐
- bash - find fprint 追加到文件而不是截断
- java - 如何将包含结构的嵌套枚举从 Swift 转换为 Java?
- python - Azure VM:如何将启动/停止/重新启动事件发送到 Webapp
- angularjs - AngularJS更新通过控制器通过函数引用的DOM中的var
- c# - 运行 azure API 应用程序时无法加载文件或程序集
- maven - 无法检索 com.google.appengine:appengine-maven-plugin:1.9.53 的插件描述符