38 lines
953 B
Go
38 lines
953 B
Go
package main
|
||
|
||
import (
|
||
"fmt"
|
||
"github.com/Shopify/sarama"
|
||
)
|
||
|
||
// kafka sarama client
|
||
func main() {
|
||
config := sarama.NewConfig()
|
||
//tailf包使用
|
||
config.Producer.RequiredAcks = sarama.WaitForAll //发送完包需要 leader和follower都确认
|
||
config.Producer.Partitioner = sarama.NewRandomPartitioner //新选出一个partition
|
||
config.Producer.Return.Successes = true
|
||
//成功交付的消息将在success channel返回
|
||
|
||
//构造一个消息
|
||
msg := &sarama.ProducerMessage{}
|
||
msg.Topic = "web_log"
|
||
msg.Value = sarama.StringEncoder("this is test log")
|
||
|
||
//连接kafka
|
||
client, err := sarama.NewSyncProducer([]string{"152.136.226.203:9093"}, config)
|
||
if err != nil {
|
||
fmt.Println("producer closed, err:", err)
|
||
return
|
||
}
|
||
fmt.Printf("连接kafka成功！\n")
|
||
defer client.Close()
|
||
|
||
pid, offset, err := client.SendMessage(msg)
|
||
if err != nil {
|
||
fmt.Println("send message err:", err)
|
||
}
|
||
fmt.Printf("pid:%v offset:%v\n", pid, offset)
|
||
|
||
}
|