关于kafka:Go-操作kafka包sarama

概述

sarama 是一个纯 Go 客户端库，用于解决 Apache Kafka（0.8 及更高版本）。它包含一个用于轻松生成和应用音讯的高级 API，以及一个用于在高级 API 有余时控制线路上的字节的低级 API。
在github上stars上比拟多（举荐）。

开源包：https://github.com/Shopify/sa...
文档地址：https://pkg.go.dev/github.com...

闲话少叙，上示例

package mainimport (    "context"    "fmt"    "github.com/Shopify/sarama"    "log"    "os"    "os/signal"    "sync")type consumerGroupHandler struct {}func (consumerGroupHandler) Setup(sarama.ConsumerGroupSession) error {    return nil}func (consumerGroupHandler) Cleanup(sarama.ConsumerGroupSession) error { return nil }// ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().func (consumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {    for msg := range claim.Messages() {        log.Printf("Message claimed: value = %s, timestamp = %v, topic = %s", msg.Value, msg.Timestamp, msg.Topic)        session.MarkMessage(msg, "")    }    return nil}//消费者组func SaramaConsumerGroup() {    config := sarama.NewConfig()    config.Consumer.Return.Errors = false    config.Version = sarama.V0_10_2_0                     // specify appropriate version    config.Consumer.Offsets.Initial = sarama.OffsetOldest // 未找到组生产位移的时候从哪边开始生产    group, err := sarama.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)        }    }()    fmt.Println("Consumed start")    // Iterate over consumer sessions.    ctx := context.Background()    for {        topics := []string{"my_topic"}        handler := consumerGroupHandler{}        // `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)        }    }}//消费者func SaramaConsumer() {    consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, sarama.NewConfig())    if err != nil {        panic(err)    }    defer func() {        if err := consumer.Close(); err != nil {            log.Fatalln(err)        }    }()    partitionConsumer, err := consumer.ConsumePartition("my_topic", 0, sarama.OffsetNewest)    if err != nil {        panic(err)    }    defer func() {        if err := partitionConsumer.Close(); err != nil {            log.Fatalln(err)        }    }()    // Trap SIGINT to trigger a shutdown.    signals := make(chan os.Signal, 1)    signal.Notify(signals, os.Interrupt)    consumed := 0ConsumerLoop:    for {        select {        case msg := <-partitionConsumer.Messages():            log.Printf("Consumed message offset %d\n", msg.Offset)            consumed++        case <-signals:            break ConsumerLoop        }    }    log.Printf("Consumed: %d\n", consumed)}//异步生产者Goroutinesfunc SyncProducer() {    config := sarama.NewConfig()    config.Producer.Return.Successes = true    config.Producer.Return.Errors = true    producer, err := sarama.NewAsyncProducer([]string{"localhost:9092"}, config)    if err != nil {        panic(err)    }    // Trap SIGINT to trigger a graceful shutdown.    signals := make(chan os.Signal, 1)    signal.Notify(signals, os.Interrupt)    var (        wg                                  sync.WaitGroup        enqueued, successes, producerErrors int    )    wg.Add(1)    go func() {        defer wg.Done()        for range producer.Successes() {            successes++        }    }()    wg.Add(1)    go func() {        defer wg.Done()        for err := range producer.Errors() {            log.Println(err)            producerErrors++        }    }()ProducerLoop:    for {        message := &sarama.ProducerMessage{Topic: "my_topic", Value: sarama.StringEncoder("testing 456")}        select {        case producer.Input() <- message:            enqueued++        case <-signals:            producer.AsyncClose() // Trigger a shutdown of the producer.            break ProducerLoop        }    }    wg.Wait()    log.Printf("Successfully produced: %d; errors: %d\n", successes, producerErrors)}//异步生产者Selectfunc SyncProducerSelect() {    producer, err := sarama.NewAsyncProducer([]string{"localhost:9092"}, nil)    if err != nil {        panic(err)    }    defer func() {        if err := producer.Close(); err != nil {            log.Fatalln(err)        }    }()    // Trap SIGINT to trigger a shutdown.    signals := make(chan os.Signal, 1)    signal.Notify(signals, os.Interrupt)    var enqueued, producerErrors intProducerLoop:    for {        select {        case producer.Input() <- &sarama.ProducerMessage{Topic: "my_topic", Key: nil, Value: sarama.StringEncoder("testing 123")}:            enqueued++        case err := <-producer.Errors():            log.Println("Failed to produce message", err)            producerErrors++        case <-signals:            break ProducerLoop        }    }    log.Printf("Enqueued: %d; errors: %d\n", enqueued, producerErrors)}//同步生产模式func SaramaProducer() {    producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, nil)    if err != nil {        log.Fatalln(err)    }    defer func() {        if err := producer.Close(); err != nil {            log.Fatalln(err)        }    }()    msg := &sarama.ProducerMessage{Topic: "my_topic", Value: sarama.StringEncoder("testing 123")}    partition, offset, err := producer.SendMessage(msg)    if err != nil {        log.Printf("FAILED to send message: %s\n", err)    } else {        log.Printf("> message sent to partition %d at offset %d\n", partition, offset)    }}func main() {    //生产者    go SyncProducer()    //go SaramaProducer()    //go SyncProducerSelect()    //消费者    SaramaConsumerGroup()    //SaramaConsumer()}

links

https://pkg.go.dev/github.com/Shopify/saramahttps://github.com/Shopify/sarama