golang如何使用sarama访问kafka
更新时间：2018年12月17日 09:21:35 作者：CodingCode
这篇文章主要介绍了golang如何使用sarama访问kafka，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
下面一个客户端代码例子访问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如何读取单行超长文本的相关资料,文中通过实例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下2021-12-12
Golang性能提升利器之SectionReader的用法详解
本文将介绍 Go 语言中的 SectionReader，包括 SectionReader的基本使用方法、实现原理、使用注意事项，感兴趣的小伙伴可以了解一下2023-07-07
这篇文章主要为大家介绍了Go应用正确的CPU核数分析，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪2023-06-06
这篇文章主要为大家介绍了Go语言如何通过通信共享内存实现示例详解,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪2023-11-11
xorm是go语言的常用orm之一，可以用来操作数据库。本文就来和大家聊聊Go语言如何使用xorm实现读取mysql功能，感兴趣的小伙伴可以跟随小编一起学习一下2022-11-11
使用Go module和GoLand初始化一个Go项目的方法
这篇文章主要介绍了使用Go module和GoLand初始化一个Go项目,本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下2020-12-12
这篇文章主要介绍了快速排序算法的原理及其Go语言版实现,文中对于快速算法的过程和效率有较为详细的说明,需要的朋友可以参考下2016-04-04
这篇文章主要为大家介绍了一文详解Golang使用接口支持Apply方法的配置模式,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪2024-01-01
本文主要介绍了Go函数使用,包括函数定义、函数声明、函数调用、可变参数函数、匿名函数、递归函数、高阶函数等,感兴趣的可以了解一下2023-11-11
SM4加密算法在安全性、高效性、简单性、标准化和广泛支持等方面具有优势，适用于各种数据保护和加密应用场景，这篇文章就来和大家探索一下Golang中的SM4加密解密算法吧2023-06-06
最新评论