Kafka 安裝配置、更多資料請(qǐng)參考其官網(wǎng)膝捞。
啟動(dòng) kafka server
在這之前需要啟動(dòng) zookeeper 做服務(wù)治理（單機(jī)）锄奢。
$ bin/zkServer.sh status conf/zoo_sample.cfg
如提示權(quán)限限制加上 sudo 搔体。
啟動(dòng) kafka server
$ bin/kafka-server-start.sh config/server.properties
啟動(dòng)消息隊(duì)列(本部分僅為測(cè)試 server)
新建 Topic
$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
$ bin/kafka-topics.sh --list --zookeeper localhost:2181 (test)
1. 啟動(dòng) Producer
$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
2. 啟動(dòng) Consumer
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
此時(shí)在 Producer 端發(fā)送消息季希，在 Consumer 就會(huì)顯示褪那，如下圖所示。
（上圖中 Consumer 多出了好幾個(gè)消息是我截圖之前測(cè)試發(fā)出的）
Action
本文使用 sarama 庫(kù)作為 kafka 的 go API胖眷。sarama 庫(kù)沒有給出很具體的文檔武通，可以參考其源碼霹崎。
Producer
package main
import (
"fmt"
"github.com/Shopify/sarama"
)
func main() {
config := sarama.NewConfig()
config.Producer.RequiredAcks = sarama.WaitForAll
config.Producer.Partitioner = sarama.NewRandomPartitioner
config.Producer.Return.Successes = true
addr := []string{"localhost:9092"}
producer, err := sarama.NewSyncProducer(addr, config)
if err != nil {
panic(err)
}
defer producer.Close()
msg := &sarama.ProducerMessage{
Topic: "hello",
Partition: int32(-1),
Key: sarama.StringEncoder("key"),
}
var value string
for {
_, err := fmt.Scanf("%s", &value)
if err != nil {
break
}
msg.Value = sarama.ByteEncoder(value)
fmt.Println(value)
partition, offset, err := producer.SendMessage(msg)
if err != nil {
fmt.Println("Send message Fail")
}
fmt.Printf("Partition = %d, offset=%d\n", partition, offset)
}
}
Consumer
package main
import (
"fmt"
"sync"
"github.com/Shopify/sarama"
)
var (
wg sync.WaitGroup
)
func main() {
consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, nil)
if err != nil {
panic(err)
}
partitionList, err := consumer.Partitions("hello")
if err != nil {
panic(err)
}
for partition := range partitionList {
pc, err := consumer.ConsumePartition("hello", int32(partition), sarama.OffsetNewest)
if err != nil {
panic(err)
}
defer pc.AsyncClose()
wg.Add(1)
go func(sarama.PartitionConsumer) {
defer wg.Done()
for msg := range pc.Messages() {
fmt.Printf("Partition:%d, Offset:%d, Key:%s, Value:%s\n", msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
}
}(pc)
}
wg.Wait()
consumer.Close()
}
結(jié)果如下：
