go的kafka生产和消费

公众号-利志分享

发布于 2022-04-25 09:15:00

2K0

文章被收录于专栏：利志分享利志分享

之前有一篇文件聊了聊如何生产不丢失数据，消费不丢失数据。这一篇我们来看下go如何通过参数配置来处理生产和消费的。

关于go的client，官方推荐有个如下几个：

https://github.com/Shopify/sarama

https://github.com/stealthly/go_kafka_client

https://github.com/stealthly/siesta

https://github.com/optiopay/kafka

https://github.com/nuance/kafka

我这里使用的就是官方推荐的第一个sarama，目前star的量7.2

go的生产端：关于生产端上篇文章也说过最核心的参数是：

  // 发送完数据需要leader和follow都确认
  mqConfig.Producer.RequiredAcks = sarama.WaitForAll

关于go的生产端核心就是确保写入的数据都到leader和follow。代码参考如下：

package main

import (
  "fmt"
  "github.com/Shopify/sarama"
  "strings"
  "time"
)

var KafkaProducer sarama.SyncProducer

func InitKafkaProducer(addressList string) {
  var err error
  mqConfig := sarama.NewConfig()
  // 设置producer
  // 发送完数据需要leader和follow都确认
  mqConfig.Producer.RequiredAcks = sarama.WaitForAll
  // Partition选择随机
  mqConfig.Producer.Partitioner = sarama.NewRandomPartitioner
  // 成功交付的消息将在success channel返回
  mqConfig.Producer.Return.Successes = true
  // 配置版本
  mqConfig.Version = sarama.V0_10_2_1
  kafkaClient, err := sarama.NewClient(strings.Split(addressList, ","), mqConfig)
  if err != nil {
    panic(err)
  }
  // 客户端生产者
  KafkaProducer, err = sarama.NewSyncProducerFromClient(kafkaClient)
  if err != nil {
    panic(err)
  }
}

func sendKafkaMessage(content, topic string) {
  msg := &sarama.ProducerMessage{
    Topic:     topic,
    Value:     sarama.StringEncoder(content),
    Timestamp: time.Now(),
  }
  partition, offset, err := KafkaProducer.SendMessage(msg)
  if err != nil {
    return
  }
  fmt.Printf("sendKafkaMessage", topic, partition, offset)
}

func main() {
  addressList := "127.0.0.1:9092"
  InitKafkaProducer(addressList)
  content := "sdfsdfrer"
  topic := "test"
  sendKafkaMessage(content, topic)
}

关于go的消费端就是手动提交offset，我们要保证写入数据库，保存成功之后提交offset，这样数据肯定是不会存在丢失的情况。下面我们来看下实现的代码：

package main

import (
  "fmt"
  "github.com/Shopify/sarama"
  cluster "github.com/bsm/sarama-cluster"
  "strings"
  "time"
)

var consumerClient *cluster.Consumer

// 初始化kafka配置消费信息
func InitKafkaConsumer(addressList string, consumerGroup string, topicList []string) {
  var err error
  clusterCfg := cluster.NewConfig()

  clusterCfg.Consumer.Return.Errors = true
  clusterCfg.Consumer.Offsets.Initial = sarama.OffsetOldest
  clusterCfg.Consumer.Offsets.AutoCommit.Enable = false
  clusterCfg.Group.Return.Notifications = true
  //这行代码是因为github.com/bsm/sarama-cluster不维护了，库会有bug，我这里只是写个demo，将就着用
  clusterCfg.Consumer.Offsets.CommitInterval = 1 * time.Second
  clusterCfg.Version = sarama.V0_10_2_1
  // kafka消费端
  if err = clusterCfg.Validate(); err != nil {
    msg := fmt.Sprintf("Kafka consumer config invalidate. config: %v. err: %v", *clusterCfg, err)
    panic(msg)
  }
  consumerClient, err = cluster.NewConsumer(strings.Split(addressList, ","), consumerGroup, topicList, clusterCfg)
  if err != nil {
    msg := fmt.Sprintf("Create kafka consumer error: %v. config: %v", err, clusterCfg)
    panic(msg)
  }

}

func main() {
  addressList := "127.0.0.1:9092"
  consumerGroup := "my_go"
  topicList := []string{"test"}
  InitKafkaConsumer(addressList, consumerGroup, topicList)
  go func() {
    for {
      select {
      case msg, ok := <-consumerClient.Messages():
        if ok {
          //这里目前我是打印数据，线上可能是写入数据到db，写入成功之后我们手动提交offset
          fmt.Println(string(msg.Value))
          //这里是临时保存offset
          consumerClient.MarkOffset(msg, "")
          //手动提交offset
          consumerClient.CommitOffsets()
        }
      case err, more := <-consumerClient.Errors():
        if more {
          fmt.Printf("consumerClient error %v\n", err)
        }
      case ntf, more := <-consumerClient.Notifications():
        if more {
          fmt.Printf("consumerClient Notifications %v \n", ntf)
        }
      }
    }
  }()
  time.Sleep(10 * time.Hour)
}

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2021-05-01，如有侵权请联系 cloudcommunity@tencent.com 删除