golang使用sarama訪問kafka的方法

小編給大家分享一下golang使用sarama訪問kafka的方法，希望大家閱讀完這篇文章之后都有所收獲，下面讓我們一起去探討吧！

十余年的錯那網(wǎng)站建設(shè)經(jīng)驗，針對設(shè)計、前端、開發(fā)、售后、文案、推廣等六對一服務(wù)，響應(yīng)快，48小時及時工作處理。成都全網(wǎng)營銷的優(yōu)勢是能夠根據(jù)用戶設(shè)備顯示端的尺寸不同，自動調(diào)整錯那建站的顯示方式，使網(wǎng)站能夠適用不同顯示終端，在瀏覽器中調(diào)整網(wǎng)站的寬度，無論在任何一種瀏覽器上瀏覽網(wǎng)站，都能展現(xiàn)優(yōu)雅布局與設(shè)計，從而大程度地提升瀏覽體驗。創(chuàng)新互聯(lián)從事“錯那網(wǎng)站設(shè)計”,“錯那網(wǎng)站推廣”以來，每個客戶項目都認(rèn)真落實執(zhí)行。

訪問kafka服務(wù)器，來發(fā)送和接受消息。

使用方式

1、命令行參數(shù)

$ ./kafkaclient -h
Usage of ./client:
 -ca string
  CA Certificate (default "ca.pem")
 -cert string
  Client Certificate (default "cert.pem")
 -command string
  consumer|producer (default "consumer")
 -host string
  Common separated kafka hosts (default "localhost:9093")
 -key string
  Client Key (default "key.pem")
 -partition int
  Kafka topic partition
 -tls
  TLS enable
 -topic string
  Kafka topic (default "test--topic")

2、作為producer啟動

$ ./kafkaclient -command producer \
 -host kafka1:9092,kafka2:9092

## TLS-enabled
$ ./kafkaclient -command producer \
 -tls -cert client.pem -key client.key -ca ca.pem \
 -host kafka1:9093,kafka2:9093

producer發(fā)送消息給kafka：

> aaa
2018/12/15 07:11:21 Produced message: [aaa]
> bbb
2018/12/15 07:11:30 Produced message: [bbb]
> quit

3、作為consumer啟動

$ ./kafkaclient -command consumer \
 -host kafka1:9092,kafka2:9092

## TLS-enabled
$ ./kafkaclient -command consumer \
 -tls -cert client.pem -key client.key -ca ca.pem \
 -host kafka1:9093,kafka2:9093

consumer從kafka接受消息：

2018/12/15 07:11:21 Consumed message: [aaa], offset: [4]
2018/12/15 07:11:30 Consumed message: [bbb], offset: [5]

完整源代碼如下

這個代碼使用到了Shopify/sarama庫，請自行下載使用。

$ cat kafkaclient.go
package main

import (
 "flag"
 "fmt"
 "log"
 "os"
 "io/ioutil"
 "bufio"
 "strings"

 "crypto/tls"
 "crypto/x509"

 "github.com/Shopify/sarama"
)

var (
 command  string
 tlsEnable bool
 hosts  string
 topic  string
 partition int
 clientcert string
 clientkey string
 cacert  string
)

func main() {
 flag.StringVar(&command, "command",  "consumer",   "consumer|producer")
 flag.BoolVar(&tlsEnable, "tls",   false,    "TLS enable")
 flag.StringVar(&hosts,  "host",   "localhost:9093", "Common separated kafka hosts")
 flag.StringVar(&topic,  "topic",  "test--topic",  "Kafka topic")
 flag.IntVar(&partition,  "partition", 0,     "Kafka topic partition")
 flag.StringVar(&clientcert, "cert",   "cert.pem",   "Client Certificate")
 flag.StringVar(&clientkey, "key",   "key.pem",   "Client Key")
 flag.StringVar(&cacert,  "ca",   "ca.pem",   "CA Certificate")
 flag.Parse()

 config := sarama.NewConfig()
 if tlsEnable {
  //sarama.Logger = log.New(os.Stdout, "[sarama] ", log.LstdFlags)
  tlsConfig, err := genTLSConfig(clientcert, clientkey, cacert)
  if err != nil {
   log.Fatal(err)
  }

  config.Net.TLS.Enable = true
  config.Net.TLS.Config = tlsConfig
 }
 client, err := sarama.NewClient(strings.Split(hosts, ","), config)
 if err != nil {
  log.Fatalf("unable to create kafka client: %q", err)
 }

 if command == "consumer" {
  consumer, err := sarama.NewConsumerFromClient(client)
  if err != nil {
   log.Fatal(err)
  }
  defer consumer.Close()
  loopConsumer(consumer, topic, partition)
 } else {
  producer, err := sarama.NewAsyncProducerFromClient(client)
  if err != nil {
   log.Fatal(err)
  }
  defer producer.Close()
  loopProducer(producer, topic, partition)
 }
}

func genTLSConfig(clientcertfile, clientkeyfile, cacertfile string) (*tls.Config, error) {
 // load client cert
 clientcert, err := tls.LoadX509KeyPair(clientcertfile, clientkeyfile)
 if err != nil {
  return nil, err
 }

 // load ca cert pool
 cacert, err := ioutil.ReadFile(cacertfile)
 if err != nil {
  return nil, err
 }
 cacertpool := x509.NewCertPool()
 cacertpool.AppendCertsFromPEM(cacert)

 // generate tlcconfig
 tlsConfig := tls.Config{}
 tlsConfig.RootCAs = cacertpool
 tlsConfig.Certificates = []tls.Certificate{clientcert}
 tlsConfig.BuildNameToCertificate()
 // tlsConfig.InsecureSkipVerify = true // This can be used on test server if domain does not match cert:
 return &tlsConfig, err
}

func loopProducer(producer sarama.AsyncProducer, topic string, partition int) {
 scanner := bufio.NewScanner(os.Stdin)
 fmt.Print("> ")
 for scanner.Scan() {
  text := scanner.Text()
  if text == "" {
  } else if text == "exit" || text == "quit" {
   break
  } else {
   producer.Input() <- &sarama.ProducerMessage{Topic: topic, Key: nil, Value: sarama.StringEncoder(text)}
   log.Printf("Produced message: [%s]\n",text)
  }
  fmt.Print("> ")
 }
}

func loopConsumer(consumer sarama.Consumer, topic string, partition int) {
 partitionConsumer, err := consumer.ConsumePartition(topic, int32(partition), sarama.OffsetNewest)
 if err != nil {
  log.Println(err)
  return
 }
 defer partitionConsumer.Close()

 for {
  msg := <-partitionConsumer.Messages()
  log.Printf("Consumed message: [%s], offset: [%d]\n", msg.Value, msg.Offset)
 }
}

編譯：

$ go build kafkaclient.go

看完了這篇文章，相信你對“golang使用sarama訪問kafka的方法”有了一定的了解，如果想了解更多相關(guān)知識，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，感謝各位的閱讀！

分享標(biāo)題：golang使用sarama訪問kafka的方法
標(biāo)題網(wǎng)址：http://bm7419.com/article14/jdccge.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供動態(tài)網(wǎng)站、商城網(wǎng)站、品牌網(wǎng)站制作、做網(wǎng)站、手機網(wǎng)站建設(shè)、Google

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)