概述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()}linkshttps://pkg.go.dev/github.com/Shopify/saramahttps://github.com/Shopify/sarama目录上一节：下一节：