我正在尝试让一些使用者来处理来自kafka的消息,并且我希望实现kubernetes部署可伸缩性以实现弹性消息处理能力。
我从萨拉玛官方指南找到了这个代码https://pkg.go.dev/github.com/shopify/sarama#newconsumergroup:
package main
import (
"context"
"fmt"
)
type exampleConsumerGroupHandler struct{}
func (exampleConsumerGroupHandler) Setup(_ ConsumerGroupSession) error { return nil }
func (exampleConsumerGroupHandler) Cleanup(_ ConsumerGroupSession) error { return nil }
func (h exampleConsumerGroupHandler) ConsumeClaim(sess ConsumerGroupSession, claim ConsumerGroupClaim) error {
for msg := range claim.Messages() {
fmt.Printf("Message topic:%q partition:%d offset:%d\n", msg.Topic, msg.Partition, msg.Offset)
sess.MarkMessage(msg, "")
}
return nil
}
func main() {
config := NewTestConfig()
config.Version = V2_0_0_0 // specify appropriate version
config.Consumer.Return.Errors = true
group, err := NewConsumerGroup([]string{"localhost:9092"}, "my-group", config)
if err != nil {
panic(err)
}
defer func() { _ = group.Close() }()
// Track errors
go func() {
for err := range group.Errors() {
fmt.Println("ERROR", err)
}
}()
// Iterate over consumer sessions.
ctx := context.Background()
for {
topics := []string{"my-topic"}
handler := exampleConsumerGroupHandler{}
// `Consume` should be called inside an infinite loop, when a
// server-side rebalance happens, the consumer session will need to be
// recreated to get the new claims
err := group.Consume(ctx, topics, handler)
if err != nil {
panic(err)
}
}
}
我有一些问题:
如何设置消费群体中的消费者数量?
如果我把这个程序部署在吊舱里,我能安全地扩展它吗?我的意思是,假设一个程序正在运行,我将复制副本从1扩展到2,会不会另一个 NewConsumerGroup
使用相同组id的呼叫可以完美地工作而不发生冲突?
先谢谢你。
注意:我使用的是kafka2.8,听说sarama\u集群包已经过时了。
1条答案
按热度按时间wvmv3b1j1#
提醒组不能扩展到主题分区计数之外
扩展pods是使用使用者组的正确方法,使用相同的组名也是正确的,但是我建议将pods和代理地址提取到环境变量中,以便在部署时可以轻松地更改它们
容器化代码将无法使用localhost作为kafka连接字符串,因为这将是pod本身