golang怎么連接kafka

這篇文章主要介紹“golang怎么連接kafka”，在日常操作中，相信很多人在golang怎么連接kafka問(wèn)題上存在疑惑，小編查閱了各式資料，整理出簡(jiǎn)單好用的操作方法，希望對(duì)大家解答”golang怎么連接kafka”的疑惑有所幫助！接下來(lái)，請(qǐng)跟著小編一起來(lái)學(xué)習(xí)吧！

1 下載，配置，啟動(dòng) kafka

配置修改

在config目錄下的server文件和zookeeper文件，其中分別修改kafka的日志保存路徑和zookeeper的數(shù)據(jù)保存路徑。

啟動(dòng)kafka

先啟動(dòng)kafka自帶的zookeeper，在kafka的根目錄下打開(kāi)終端，使用配置文件啟動(dòng)

./bin/windows/zookeeper-server-start.bat config/zookeeper.properties

同樣在kafka目錄的根目錄下啟動(dòng)kafka

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

2 使用golang的github.com/Shopify/sarama庫(kù)連接kafka

package main

import (
	"fmt"
	"time"

	"github.com/Shopify/sarama"
)

func main() {
	config:=sarama.NewConfig()
	// 生產(chǎn)者配置
	config.Producer.RequiredAcks=sarama.WaitForAll
	config.Producer.Partitioner=sarama.NewRandomPartitioner
	config.Producer.Return.Successes=true
	// 封裝消息
	msg:=&sarama.ProducerMessage{}
	msg.Topic="shopping"
	time_str:=time.Now().Format("2006-01-02 15:04:05")
	msg.Value=sarama.StringEncoder("0413 test log!"+time_str)
	// 連接kafka
	client,err:=sarama.NewSyncProducer([]string{"127.0.0.1:9092"}, config)
	if err!=nil {
		fmt.Println("producer closed", err)
		return
	}
	defer client.Close()
	// 發(fā)送消息
	partition,offset,err:=client.SendMessage(msg)
	if err!=nil {
		fmt.Println("send failed", err)
		return
	}
	fmt.Printf("partition:%v offset:%v", partition, offset)
}

這段代碼實(shí)現(xiàn)了模擬生產(chǎn)者向kafka發(fā)送消息的過(guò)程，包含：配置生產(chǎn)者，封裝消息，消息類(lèi)型是 *sarama.ProducerMessage，連接kafka，默認(rèn)端口是9092，發(fā)送消息，返回消息存儲(chǔ)的partition和offset日志偏移量。

3 確認(rèn)生產(chǎn)者發(fā)送成功

使用kafka自帶的命令行消費(fèi)者客戶(hù)端查看kafka中的數(shù)據(jù)
在kafka的根目錄下

bin/windows/kafka-console-consumer.bat --bootstrap-server 127.0.0.1:9092 --topic shopping --from-beginning

這里的topic和代碼中的topic一致，均為shopping
終端會(huì)輸出之前發(fā)送的數(shù)據(jù)。

到此，關(guān)于“golang怎么連接kafka”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實(shí)踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識(shí)，請(qǐng)繼續(xù)關(guān)注億速云網(wǎng)站，小編會(huì)繼續(xù)努力為大家?guī)?lái)更多實(shí)用的文章！