Golang操作Kafka的實現(xiàn)示例

更新時間：2023年02月19日 09:00:04 作者：YUHAOHAO

本文主要介紹了Golang操作Kafka的實現(xiàn)示例，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧

一.使用庫說明

Golang中連接kafka可以使用第三方庫：github.com/Shopify/sarama

二.Kafka Producer發(fā)送消息

package main?

import (
?? ?"fmt"
?? ?"github.com/Shopify/sarama"
)

func main() {
?? ?config := sarama.NewConfig()
?? ?config.Producer.RequiredAcks = sarama.WaitForAll // 發(fā)送完數(shù)據(jù)需要leader和follower都確認
?? ?config.Producer.Partitioner = sarama.NewRandomPartitioner ?//寫到隨機分區(qū)中，我們默認設置32個分區(qū)
?? ?config.Producer.Return.Successes = true // 成功交付的消息將在success channel返回

?? ?// 構造一個消息
?? ?msg := &sarama.ProducerMessage{}
?? ?msg.Topic = "task"
?? ?msg.Value = sarama.StringEncoder("producer kafka messages...")

?? ?// 連接kafka
?? ?client, err := sarama.NewSyncProducer([]string{"192.20.216.8:9092"}, config)
?? ?if err != nil {
?? ??? ?fmt.Println("Producer closed, err:", err)
?? ??? ?return
?? ?}
?? ?defer client.Close()

?? ?// 發(fā)送消息
?? ?pid, offset, err := client.SendMessage(msg)
?? ?if err != nil {
?? ??? ?fmt.Println("send msg failed, err:", err)
?? ??? ?return
?? ?}
?? ?fmt.Printf("pid:%v offset:%v\n", pid, offset)
}

三.Kafka Consumer消費消息

package main

import (
?? ?"fmt"
?? ?"github.com/Shopify/sarama"
?? ?"sync"
)

func main() {
?? ?var wg sync.WaitGroup
?? ?consumer, err := sarama.NewConsumer([]string{"192.20.216.8:9092"}, nil)
?? ?if err != nil {
?? ??? ?fmt.Println("Failed to start consumer: %s", err)
?? ??? ?return
?? ?}
?? ?partitionList, err := consumer.Partitions("task-status-data") // 通過topic獲取到所有的分區(qū)
?? ?if err != nil {
?? ??? ?fmt.Println("Failed to get the list of partition: ", err)
?? ??? ?return
?? ?}
?? ?fmt.Println(partitionList)

?? ?for partition := range partitionList{ // 遍歷所有的分區(qū)
?? ??? ?pc, err := consumer.ConsumePartition("task", int32(partition), sarama.OffsetNewest) // 針對每個分區(qū)創(chuàng)建一個分區(qū)消費者
?? ??? ?if err != nil {
?? ??? ??? ?fmt.Println("Failed to start consumer for partition %d: %s\n", partition, err)
?? ??? ?}
?? ??? ?wg.Add(1)
?? ??? ?go func(sarama.PartitionConsumer) { // 為每個分區(qū)開一個go協(xié)程取值
?? ??? ??? ?for msg := range pc.Messages() { // 阻塞直到有值發(fā)送過來，然后再繼續(xù)等待
?? ??? ??? ??? ?fmt.Printf("Partition:%d, Offset:%d, key:%s, value:%s\n", msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
?? ??? ??? ?}
?? ??? ??? ?defer pc.AsyncClose()
?? ??? ??? ?wg.Done()
?? ??? ?}(pc)
?? ?}
?? ?wg.Wait()
?? ?consumer.Close()
}

到此這篇關于Golang操作Kafka的實現(xiàn)示例的文章就介紹到這了,更多相關Golang操作Kafka內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: