|
|
package main
|
|
|
|
|
|
import (
|
|
|
"fmt"
|
|
|
"github.com/Shopify/sarama"
|
|
|
"hash"
|
|
|
"hash/crc32"
|
|
|
"log"
|
|
|
"math/rand"
|
|
|
"os"
|
|
|
"os/signal"
|
|
|
"sync"
|
|
|
"time"
|
|
|
)
|
|
|
|
|
|
func main() {
|
|
|
// 一，得到异步的producer
|
|
|
brokers := []string{"localhost:9092"}
|
|
|
conf := sarama.NewConfig()
|
|
|
// 开启 Success channel 来接收发送成功的信息
|
|
|
conf.Producer.Return.Successes = true
|
|
|
// 配置producer选项分区的策略
|
|
|
// 随机，全部的消息不需要逻辑上的分割
|
|
|
//conf.Producer.Partitioner = sarama.NewRandomPartitioner
|
|
|
// 轮循,全部的消息不需要逻辑上的分割
|
|
|
//conf.Producer.Partitioner = sarama.NewRoundRobinPartitioner
|
|
|
|
|
|
// Hash，基于特定的Key，完成分区选择。具有逻辑含义
|
|
|
//conf.Producer.Partitioner = sarama.NewHashPartitioner
|
|
|
//自定义Hash算法，需要完成hash32的函数
|
|
|
conf.Producer.Partitioner = sarama.NewCustomHashPartitioner(func() hash.Hash32 {
|
|
|
//poly: x³²+ x³¹+ x²⁴+ x²²+ x¹⁶+ x¹⁴+ x⁸+ x⁷+ x⁵
|
|
|
//0b11010101100000101000001010000001
|
|
|
return crc32.New(crc32.MakeTable(0xD5828281))
|
|
|
})
|
|
|
// 指定，配合 partition使用
|
|
|
//conf.Producer.Partitioner = sarama.NewManualPartitioner
|
|
|
producer, err := sarama.NewAsyncProducer(brokers, conf)
|
|
|
if err != nil {
|
|
|
log.Fatalln(err)
|
|
|
}
|
|
|
|
|
|
// 二，启用 goroutine
|
|
|
var wg sync.WaitGroup
|
|
|
sendCounter, errorCounter, successCounter := 0, 0, 0
|
|
|
|
|
|
// 2.1 处理 errors
|
|
|
go func() {
|
|
|
wg.Add(1)
|
|
|
defer wg.Done()
|
|
|
for err := range producer.Errors() {
|
|
|
log.Printf("Failed to send, err: %s\n", err)
|
|
|
errorCounter++
|
|
|
}
|
|
|
}()
|
|
|
|
|
|
// 2.2 处理 success
|
|
|
go func() {
|
|
|
wg.Add(1)
|
|
|
defer wg.Done()
|
|
|
for suc := range producer.Successes() {
|
|
|
log.Printf("Success to send, partition: %d, offset: %d, value: %s\n", suc.Partition, suc.Offset, suc.Value)
|
|
|
successCounter++
|
|
|
}
|
|
|
}()
|
|
|
|
|
|
// 增加控制信号
|
|
|
signals := make(chan os.Signal, 1)
|
|
|
signal.Notify(signals, os.Interrupt) // ctrl + c
|
|
|
|
|
|
loop:
|
|
|
|
|
|
for {
|
|
|
// 生成特定的消息,随机生成
|
|
|
// 消息关联的数据id
|
|
|
id := rand.Intn(5)
|
|
|
|
|
|
time.Sleep(1000 * time.Millisecond)
|
|
|
select {
|
|
|
case producer.Input() <- &sarama.ProducerMessage{
|
|
|
Topic: "topic_more_partition_1", // 3 partitions
|
|
|
Value: sarama.StringEncoder("MaShiBing Go, id:" + fmt.Sprintf("%d", id)),
|
|
|
// Hash
|
|
|
//Key: sarama.StringEncoder(fmt.Sprintf("%d", id)),
|
|
|
// manual
|
|
|
Partition: int32(id % 3),
|
|
|
}:
|
|
|
sendCounter++
|
|
|
|
|
|
// for 终止信号
|
|
|
case <-signals:
|
|
|
// 异步终止
|
|
|
producer.AsyncClose()
|
|
|
break loop
|
|
|
|
|
|
}
|
|
|
}
|
|
|
|
|
|
// wg 等待
|
|
|
wg.Wait()
|
|
|
|
|
|
// 输出统计结果
|
|
|
log.Printf("SendCounter: %d, ErrorCounter: %d, SuccessCounter: %d\n", sendCounter, errorCounter, successCounter)
|
|
|
|
|
|
}
|
