Go语言中的Kafka操作：sarama实战
1. sarama简介
sarama是Go语言中流行的Apache Kafka客户端库，提供了生产者和消费者API。本文将介绍如何使用sarama进行Kafka操作。
2. 安装和配置
go get github.com/IBM/sarama
3. 生产者
package main
import (
"github.com/IBM/sarama"
"log"
)
func main() {
config := sarama.NewConfig()
config.Producer.Return.Successes = true
producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
if err != nil {
log.Fatal(err)
}
defer producer.Close()
msg := &sarama.ProducerMessage{
Topic: "test-topic",
Value: sarama.StringEncoder("Hello Kafka"),
}
partition, offset, err := producer.SendMessage(msg)
if err != nil {
log.Fatal(err)
}
log.Printf("Message sent to partition %d at offset %d\\n", partition, offset)
}
4. 消费者
package main
import (
"github.com/IBM/sarama"
"log"
"os"
"os/signal"
)
func main() {
config := sarama.NewConfig()
config.Consumer.Return.Errors = true
consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, config)
if err != nil {
log.Fatal(err)
}
defer consumer.Close()
partitionConsumer, err := consumer.ConsumePartition("test-topic", 0, sarama.OffsetNewest)
if err != nil {
log.Fatal(err)
}
defer partitionConsumer.Close()
signals := make(chan os.Signal, 1)
signal.Notify(signals, os.Interrupt)
for {
select {
case msg := <-partitionConsumer.Messages():
log.Printf("Received message: %s\\n", string(msg.Value))
case err := <-partitionConsumer.Errors():
log.Printf("Error: %v\\n", err)
case <-signals:
return
}
}
}
5. 消费者组
package main
import (
"context"
"github.com/IBM/sarama"
"log"
)
type ConsumerGroupHandler struct{}
func (h ConsumerGroupHandler) Setup(sarama.ConsumerGroupSession) error { return nil }
func (h ConsumerGroupHandler) Cleanup(sarama.ConsumerGroupSession) error { return nil }
func (h ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for msg := range claim.Messages() {
log.Printf("Message: %s\\n", string(msg.Value))
session.MarkMessage(msg, "")
}
return nil
}
func main() {
config := sarama.NewConfig()
config.Version = sarama.V2_0_0_0
config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
group, err := sarama.NewConsumerGroup([]string{"localhost:9092"}, "my-group", config)
if err != nil {
log.Fatal(err)
}
defer group.Close()
handler := ConsumerGroupHandler{}
ctx := context.Background()
for {
err := group.Consume(ctx, []string{"test-topic"}, handler)
if err != nil {
log.Printf("Error: %v\\n", err)
}
}
}
6. 总结
使用sarama时应注意：
