golang如何使用sarama访问kafka

 更新时间:2019-04-17 21:47:19   作者:佚名   我要评论(0)

下面一个客户端代码例子访问kafka服务器,来发送和接受消息。
使用方式
1、命令行参数




$ ./kafkaclient -h
Usage of ./client:
-ca string
CA Certif

下面一个客户端代码例子访问kafka服务器,来发送和接受消息。

使用方式

1、命令行参数

$ ./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发送消息给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

    golang如何使用sarama访问kafka

    下面一个客户端代码例子访问kafka服务器,来发送和接受消息。 使用方式 1、命令行参数 $ ./kafkaclient -h Usage of ./client: -ca string CA Certif
    2019-04-17
  • Go语言下载网络图片或文件的方法示例

    Go语言下载网络图片或文件的方法示例

    最近闲来无事, 于是就简单学习了下Go语言的基本的用法。由于实践才是最快的学习方法,所以这里就以下载网络图片或文件入手来学习Go语言 文件下载到本地,
    2019-04-17
  • Golang数组的传递详解

    Golang数组的传递详解

    概念介绍 数组与切片 数组是具有相同唯一类型的一组已编号且长度固定的数据项序列。数组长度最大为2Gb,它是值类型。切片是对数组一个连续片段的引用,所以切
    2019-04-17
  • Golang中如何对MySQL进行操作详解

    Golang中如何对MySQL进行操作详解

    前言 Golang官方并没有提供数据库驱动,但通过database/sql/driver包来提供了实现驱动的标准接口。可以在Github上找到很多开源的驱动。 其中go-sql-driver
    2019-04-17
  • go json转换实践中遇到的坑

    go json转换实践中遇到的坑

    在使用 go 语言开发过程中,经常需要使用到 json 包来进行 json 和 struct 的互相转换,在使用过程中,遇到了一些需要额外注意的地方,记录如下。 整数变浮点
    2019-04-17
  • Go routine调度详解

    Go routine调度详解

    goroutine简介 goroutine是go语言中最为NB的设计,也是其魅力所在,goroutine的本质是协程,是实现并行计算的核心。goroutine使用方式非常的简单,只需使用go
    2019-04-17
  • golang高并发的深入理解

    golang高并发的深入理解

    前言 GO语言在WEB开发领域中的使用越来越广泛,Hired 发布的《2019 软件工程师状态》报告中指出,具有 Go 经验的候选人是迄今为止最具吸引力的。平均每位求
    2019-04-17
  • 详解Golang中下划线的使用方法

    详解Golang中下划线的使用方法

    在 Golang 里, _ (下划线)是个特殊的标识符。前几天看 gin 源码,看到一个有意思的用法。虽然网上的总结博客已有很多,但是总是有点欠缺,于是就有了这一篇
    2019-04-17
  • Go并发调用的超时处理的方法

    Go并发调用的超时处理的方法

    之前有聊过 golang 的协程,我发觉似乎还很理论,特别是在并发安全上,所以特结合网上的一些例子,来试验下go routine中 的 channel, select, context 的妙用
    2019-04-17
  • golang线程安全的map实现

    golang线程安全的map实现

    网上找的协程安全的map都是用互斥锁或者读写锁实现的,这里用单个协程来实现下,即所有的增删查改操作都集成到一个goroutine中,这样肯定不会出现多线程并发访
    2019-04-17

最新评论