golang連接kafka消費進ES操作

更新時間：2020年12月18日 14:24:23 作者：毛毛蟲de一生

這篇文章主要介紹了golang連接kafka消費進ES操作，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧

1.首先初始化conf配置把kafka和ES的地址配置好還有一個日志方便查看

配置信息如下用到的庫是

github.com/astaxie/beego/config
[logs]
log_level = debug
log_path = "./logs/log_transfer.log"
[kafka]
server_addr = 192.168.0.134:9092
topic = nginx_log
[ES]
addr = http://192.168.0.134:9200/

2.讀取conf配置存取進結(jié)構(gòu)體

type LogConfig struct {
  kafkaAddr string
  ESAddr string
  LogPath string
  LogLevel string
  Topic string
}
var (
   logConfig *LogConfig
)

3.讀取conf配置代碼如下

func initConfig(conftype string,filename string)(err error) {
   conf, err := config. NewConfig(conftype,filename)
   if err != nil {
    fmt. Println( "new config faild,err:",err)
     return
  }
   logConfig = &LogConfig{}
   logConfig.LogLevel = conf. String( "logs::log_level")
   if len(logConfig.LogLevel) == 0 {
     logConfig.LogLevel = "debug"
  }
   logConfig.LogPath = conf. String( "logs::log_path")
   if len(logConfig.LogPath) == 0 {
     logConfig.LogPath = "./logs"
  }
   logConfig.kafkaAddr = conf. String( "kafka::server_addr")
   if len(logConfig.kafkaAddr) == 0 {
     err = fmt. Errorf( "invalid kafka addr err")
     return
  }
   logConfig.ESAddr = conf. String( "ES::addr")
   if len(logConfig.ESAddr) == 0 {
     err = fmt. Errorf( "invalid ES addr err")
     return
  }
   logConfig.Topic = conf. String( "kafka::topic")
   if len(logConfig.Topic) == 0 {
     err = fmt. Errorf( "invalid topic addr err")
     return
  }
   return
}

4.完成了initConfig的初始化

5.初始化initLogger

func convertLogLevel(level string) int {
   switch(level) {
     case "debug":
       return logs.LevelDebug
     case "warn":
       return logs.LevelWarn
     case "info":
       return logs.LevelInfo
     case "trace":
       return logs.LevelTrace
  }
   return logs.LevelDebug
}
func initLogger(logpath string, logLevel string) (err error) {
   config := make( map[ string] interface{})
  config[ "filename"] = logpath
  config[ "level"] = convertLogLevel(logLevel)
   configStr, err := json. Marshal(config)
   if err!= nil {
    fmt. Println( "marshal failed,err:",err)
     return
  }
  logs. SetLogger(logs.AdapterFile, string(configStr))
   return
}

6.初始化kafka

type KafkaClient struct {
  client sarama.Consumer
  addr string
  topic string
  wg sync.WaitGroup
}
var (
   kafkaClient *KafkaClient
)
func initKafKa(addr string,topic string)(err error) {
   kafkaClient = &KafkaClient{}
   consumer, err := sarama. NewConsumer(strings. Split(addr, ","), nil)
   if err != nil {
    logs. Error( "Failed to strat consumer :",err)
     return 
  }
   kafkaClient.client = consumer
   kafkaClient.addr = addr
   kafkaClient.topic = topic
   return
}

7.初始化ES

gopkg.in/olivere/elastic.v2 // 這個是操作ES的庫

type LogMessage struct {
  App string
  Topic string
  Message string
}
var (
   esClient *elastic.Client
)
func initES(addr string)(err error) {
  
   client, err := elastic. NewClient(elastic. SetSniff( false),elastic. SetURL(addr))
   if err != nil {
    fmt. Println( "connect es error",err)
     return
  }
   esClient = client
   return
}

8.干活把kafka的數(shù)據(jù)寫入ES

github.com/Shopify/sarama 這個是操作kafka的驅(qū)動庫

func run()(err error) {
  fmt. Println( "run")
   partitionList, err := kafkaClient.client. Partitions(kafkaClient.topic)
   if err != nil {
    logs. Error( "ini failed ,err:%v",err)
    fmt. Printf( "ini failed ,err:%v",err)
     return
  }
   for partition := range partitionList {
    fmt. Println( "for進入")
     pc, errRet := kafkaClient.client. ConsumePartition(kafkaClient.topic, int32(partition),sarama.OffsetNewest)
     if errRet != nil {
       err = errRet
      logs. Error( "Failed to start consumer for partition %d: %s \n ",partition,err)
      fmt. Printf( "Failed to start consumer for partition %d: %s \n ",partition,err)
       return
    }
     defer pc. AsyncClose()
    fmt. Println( "馬上進入?yún)f(xié)程")
    kafkaClient.wg. Add( 1)
     go func(pc sarama.PartitionConsumer){
      fmt. Println( "進來了")
      
       for msg := range pc. Messages() {
        fmt. Println( "func執(zhí)行")
        logs. Debug( "Partition:%d,Offset:%d,key:%s,value:%s",msg.Partition,msg.Offset, string(msg.Key), string(msg.Value))
         //fmt.Println()
         err = sendToES(kafkaClient.topic,msg.Value)
         if err != nil {
          logs. Warn( "send to es failed,err:%v",err)
        }
      }
      kafkaClient.wg. Done()
    }(pc)    
  }
  kafkaClient.wg. Wait()
  fmt. Println( "協(xié)程執(zhí)行完畢")
   return
}

上面代碼是讀kafka消費數(shù)據(jù)通過sendToES這個函數(shù)發(fā)送至ES里面

sendToES代碼如下

func sendToES(topic string,data [] byte) (err error) {
   msg := &LogMessage{}
   msg.Topic = topic
   msg.Message = string(data)
   _, err = esClient. Index().
       Index(topic).
       Type(topic).
       BodyJson(msg).
       Do()
     if err != nil {
      
       return
    }
   return
}

Index就是索引名稱

index().type().bodyjson().do()這樣的寫法是鏈式執(zhí)行操作

9.寫完了基本操作后再寫一個模擬寫入數(shù)據(jù)進kafka的數(shù)據(jù) 代碼如下

func main() {
config := sarama. NewConfig()
config.Producer.RequiredAcks = sarama.WaitForAll
config.Producer.Partitioner = sarama.NewRandomPartitioner
config.Producer.Return.Successes = true
client, err := sarama. NewSyncProducer([] string{ "127.0.0.1:9092"}, config)
if err != nil {
fmt. Println( "producer close,err:", err)
return
}
defer client. Close()
var n int= 0
for {
n++
msg := &sarama.ProducerMessage{}
msg.Topic = "nginx_log"
msg.Value = sarama. StringEncoder( "this is a good test,hello maomaoChong!!," + strconv. Itoa(n))
pid, offset, err := client. SendMessage(msg)
if err != nil {
fmt. Println( "send message failed,", err)
return
}
fmt. Printf( "pid:%v offset:%v \n ", pid, offset)
time. Sleep(time.Second * 2)
}
}

這個就是生產(chǎn)者往kafka里面寫入數(shù)據(jù)進去消費

10.我們啟動我們的kafka 注意kafka依賴于zookeeper 先啟動ZK然后啟動kafka

我這里用的是zookeeper-3.4.12網(wǎng)上有下載

啟動ZK

ZK已經(jīng)成功啟動

11.啟動kafka 我這里是kafka_2.11-1.1.0

.\bin\windows\kafka-server-start.bat .\config\server.properties

kafka已經(jīng)跑起來了

12.把kafka消費測試端也啟動

.\bin\windows\kafka-console-consumer.bat --topic nginx_log --zookeeper 127.0.0.1 2181

消費端啟動成功一直等待數(shù)據(jù)進來消費

13.然后我們把ES 和Kib 都啟動了

這是我們的ES版本是5.5.1的已經(jīng)跑起來了接著啟動我們的kib

kib里面有個配置config下面的叫kibana.yml里面設置好ES的地址和端口就處于監(jiān)聽ES狀態(tài)

啟動kib有點慢稍微等一下就好

此時啟動好了kib

14.測試kib是否啟動

默認地址是http://localhost:5601

進入成功確定沒問題

15.編譯我們的代碼寫數(shù)據(jù)進kafka

從上面看我們知道一個再寫一個再消費

16.編譯運行我們把kafka寫入進ES里面的代碼

運行了這里就把kafka消費的數(shù)據(jù) 寫入進ES里面

17.我們看一下數(shù)據(jù)是否有進入kib

我們看到有數(shù)據(jù)了已經(jīng)成功了~

以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。如有錯誤或未考慮完全的地方，望不吝賜教。

您可能感興趣的文章:

Golang實現(xiàn)文件夾的創(chuàng)建與刪除的方法詳解
這篇文章主要介紹了如何利用Go語言實現(xiàn)對文件夾的常用操作：創(chuàng)建于刪除。文中的示例代碼講解詳細，感興趣的小伙伴可以跟隨小編一起學習一下
2022-05-05
Go語言對前端領域的入侵WebAssembly運行原理
這篇文章主要為大家介紹了不安分的Go語言對Web?前端領域的入侵WebAssembly運行原理實例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪
2022-07-07
一鍵定位Golang線上服務內(nèi)存泄露的秘籍
內(nèi)存泄露？別讓它拖垮你的Golang線上服務！快速掌握Go語言服務內(nèi)存泄漏排查秘籍,從此問題無處遁形,一文讀懂如何精準定位與有效解決Golang應用中的內(nèi)存問題,立即閱讀,讓性能飛升！
2024-01-01
Golang實現(xiàn)程序優(yōu)雅退出的方法詳解
項目開發(fā)過程中，隨著需求的迭代，代碼的發(fā)布會頻繁進行，在發(fā)布過程中，Golang如何讓程序做到優(yōu)雅的退出？本文就來詳細為大家講講
2022-06-06
從并發(fā)到并行解析Go語言中的sync.WaitGroup
Go?語言提供了許多工具和機制來實現(xiàn)并發(fā)編程，其中之一就是?sync.WaitGroup。本文就來深入討論?sync.WaitGroup，探索其工作原理和在實際應用中的使用方法吧
2023-05-05
重學Go語言之文件操作詳解
有很多場景都需要對文件進行讀取或者寫入，比如讀取配置文件或者寫入日志文件，在Go語言中，操作文件應該算是一件比較簡單的事情，我們在這一篇文章中，一起來探究一下
2023-08-08
go語言中的udp協(xié)議及TCP通訊實現(xiàn)示例
這篇文章主要為大家介紹了go語言中的udp協(xié)議及TCP通訊的實現(xiàn)示例，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步早日升職加薪
2022-04-04
Golang實現(xiàn)支持多種類型的set
在項目開發(fā)中，常常會用到set去重，為什么不寫一個set呢，而且go現(xiàn)在支持了泛型，所以本文就來用Golang實現(xiàn)一個支持多種類型的set呢
2023-05-05
golang?中?recover()的使用方法
這篇文章主要介紹了Guam與golang??recover()的使用方法，Recover?是一個Go語言的內(nèi)建函數(shù)，可以讓進入宕機流程中的?goroutine?恢復過來，下文更多相關資料需要的小伙伴可以參考一下
2022-04-04
golang解析html網(wǎng)頁的方法
今天小編就為大家分享一篇golang解析html網(wǎng)頁的方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧
2019-08-08