Kafka

基于confluent-kafka-go封装的kafka库

特性

• 简化Consumer实例初始化流程，基于配置自动对Consumer进行初始化且启动
• 简化Producer实例初始化流程，基于配置自动对Producer进行初始化且启动

安装

go get github.com/ant-libs-go/http

KafkaConsumer 快速开始

• toml 配置文件

  [kafka.consumer.default]
      addrs = ["127.0.0.1:9092"]
      topics = ["business"]
      group_id = "consumertest"
      consume_worker_num = 1
      receive_worker_num = 1
  

• 使用方法

   // 初始化config包，参考config模块
   code...
  
   // 如下方式可以直接使用kafka consumer实例
   err := kafka.Receive("default", func(topic string, worker int, body string, msg *kafka.Message) error {
       fmt.Println(body)
   }); 
   if err != nil {
       fmt.Printf("[ERROR] Build kafka consumer error: %s\n", err)
       os.Exit(-1)
   }
  
   // 停止kafka consumer
   kafka.CloseConsumer("default")

KafkaProducer 快速开始

• toml 配置文件

  [kafka.producer.default]
      addrs = ["127.0.0.1:9092"]
      acks = 1
      topic = "business"
      partitioner = "hash"
      return_successes = true
      return_errors = true
      return_feedback_num = 10
  

• 使用方法

   // 初始化config包，参考config模块
   code...
  
   err := kafka.Publish("default", "bodystring", "keystring", 0)
   if err == nil {
       fmt.Println(err)
   }