golang連接kafka的示例代碼
1 下載,配置,啟動 kafka
配置修改
在config目錄下的server文件和zookeeper文件,其中分別修改kafka的日志保存路徑和zookeeper的數據保存路徑。
啟動kafka
先啟動kafka自帶的zookeeper,在kafka的根目錄下打開終端,使用配置文件啟動
./bin/windows/zookeeper-server-start.bat config/zookeeper.properties
同樣在kafka目錄的根目錄下啟動kafka
./bin/windows/kafka-server-start.bat config/server.properties
2 使用golang的github.com/Shopify/sarama庫連接kafka
package main import ( "fmt" "time" "github.com/Shopify/sarama" ) func main() { config:=sarama.NewConfig() // 生產者配置 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) }
這段代碼實現了模擬生產者向kafka發(fā)送消息的過程,包含:配置生產者,封裝消息,消息類型是 *sarama.ProducerMessage
,連接kafka,默認端口是9092,發(fā)送消息,返回消息存儲的partition和offset日志偏移量。
3 確認生產者發(fā)送成功
使用kafka自帶的命令行消費者客戶端查看kafka中的數據
在kafka的根目錄下
bin/windows/kafka-console-consumer.bat --bootstrap-server 127.0.0.1:9092 --topic shopping --from-beginning
這里的topic和代碼中的topic一致,均為shopping
終端會輸出之前發(fā)送的數據。
到此這篇關于golang連接kafka的示例代碼的文章就介紹到這了,更多相關golang連接kafka內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家!
相關文章
golang的時區(qū)和神奇的time.Parse的使用方法
這篇文章主要介紹了golang的時區(qū)和神奇的time.Parse的使用方法,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2021-04-04