需要借助的库
github.com/Shopify/sarama // kafka主要的库*
github.com/bsm/sarama-cluster // kafka消费组生产者
package producer
import (
"fmt"
"github.com/HappyTeemo7569/teemoKit/tlog"
"github.com/Shopify/sarama"
"kafkaDemo/define"
)
var (
ProducerId = 1
)
type Producer struct {
Producer sarama.SyncProducer
Topic string //主题
ProducerID int //生产者Id
MessageId int
}
func (p *Producer) InitProducer() {
config := sarama.NewConfig()
config.Producer.RequiredAcks = sarama.WaitForAll // 发送完数据需要leader和follow都确认
config.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出一个partition
config.Producer.Return.Successes = true // 成功交付的消息将在success channel返回
// 连接kafka
client, err := sarama.NewSyncProducer([]string{define.SERVER_LIST}, config)
if err != nil {
tlog.Error("producer closed, err:", err)
return
}
p.Producer = client
p.Topic = define.TOPIC
p.ProducerID = ProducerId
p.MessageId = 1
ProducerId++
}
func (p *Producer) SendMessage() {
// 构造一个消息
msg := &sarama.ProducerMessage{}
msg.Topic = p.Topic
txt := fmt.Sprintf("ProducerID:%d this is a test log %d",
p.ProducerID, p.MessageId)
msg.Value = sarama.StringEncoder(txt)
// 发送消息
pid, offset, err := p.Producer.SendMessage(msg)
//_, _, err := client.SendMessage(msg)
if err != nil {
fmt.Println("send msg failed, err:", err)
return
}
tlog.Info(fmt.Sprintf("ProducerID:%d pid:%v offset:%v msg:%s",
p.ProducerID, pid, offset, txt))
p.MessageId++
}
func (p *Producer) Close() {
p.Producer.Close()
}消费者
package consumer
import (
"github.com/HappyTeemo7569/teemoKit/tlog"
"github.com/Shopify/sarama"
"kafkaDemo/define"
)
type Consumer struct {
Consumer sarama.Consumer
Topic string
ConsumerId int //消费者Id
}
func (c *Consumer) InitConsumer() error {
consumer, err := sarama.NewConsumer([]string{define.SERVER_LIST}, nil)
if err != nil {
return err
}
c.Consumer = consumer
c.Topic = define.TOPIC
c.ConsumerId = ConsumerId
ConsumerId++
return nil
}
//指定partition
//offset 可以指定，传-1为获取最新offest
func (c *Consumer) GetMessage(partitionId int32, offset int64) {
if offset == -1 {
offset = sarama.OffsetNewest
}
pc, err := c.Consumer.ConsumePartition(c.Topic, partitionId, offset)
if err != nil {
tlog.Error("failed to start consumer for partition %d,err:%v", partitionId, err)
//That topic/partition is already being consumed
return
}
// 异步从每个分区消费信息
go func(sarama.PartitionConsumer) {
for msg := range pc.Messages() {
tlog.Info("ConsumerId:%d Partition:%d Offset:%d Key:%v Value:%v", c.ConsumerId, msg.Partition, msg.Offset, msg.Key, string(msg.Value))
}
}(pc)
}
//遍历所有分区
func (c *Consumer) GetMessageToAll(offset int64) {
partitionList, err := c.Consumer.Partitions(c.Topic) // 根据topic取到所有的分区
if err != nil {
tlog.Error("fail to get list of partition:err%v", err)
return
}
tlog.Info("所有partition:", partitionList)
for partition := range partitionList { // 遍历所有的分区
c.GetMessage(int32(partition), offset)
}
}主函数
func main() {
tlog.Info("开始")
go producer.Put()
go consumer.Get()
for {
time.Sleep(time.Hour * 60)
}
}
func Put() {
producer := new(Producer)
producer.InitProducer()
go func() {
for {
producer.SendMessage()
time.Sleep(1 * time.Second)
}
}()
}
func Get() {
offest := int64(0)
consumer := new(Consumer)
err := consumer.InitConsumer()
if err != nil {
tlog.Error("fail to init consumer, err:%v", err)
return
}
consumer.GetMessageToAll(offest)
}具体源码可以查看：
kafka_demo
生产环境中的优化
- 可以存储消费的节点到redis
- 需要顺序的消费的放到一个partition，或者利用哈希算法投递
- 传入一个通道，将业务逻辑和底层逻辑解耦。
package kafka
import (
"crypto/tls"
"crypto/x509"
"fmt"
"github.com/Shopify/sarama"
"io/ioutil"
"log"
"sync"
"vliao.com/stellar/internal/core"
)
type KafkaConsumer struct {
Node []string
Consumer sarama.Consumer
Topic string
MessageQueue chan []byte
}
func NewKafkaConsumer(topic string) KafkaConsumer {
return KafkaConsumer{
Node: core.GetKafkaConn().Conn,
Topic: core.GetServerMode() + "_" + topic,
}
}
// Consume 获取所有分区
func (c *KafkaConsumer) Consume() {
config := sarama.NewConfig()
config.Net.SASL.Enable = true
config.Net.SASL.User = core.GetKafkaConn().SASLUser
config.Net.SASL.Password = core.GetKafkaConn().SASLPassword
config.Net.SASL.Handshake = true
certBytes, err := ioutil.ReadFile(GetFullPath("only-4096-ca-cert"))
if err != nil {
fmt.Println("kafka client read cert file failed ", err.Error())
return
}
clientCertPool := x509.NewCertPool()
ok := clientCertPool.AppendCertsFromPEM(certBytes)
if !ok {
fmt.Println("kafka client failed to parse root certificate")
return
}
config.Net.TLS.Config = &tls.Config{
RootCAs: clientCertPool,
InsecureSkipVerify: true,
}
config.Net.TLS.Enable = true
consumer, err := sarama.NewConsumer(c.Node, config)
if err != nil {
log.Fatal("NewConsumer err: ", err)
}
defer consumer.Close()
// 先查询该 topic 有多少分区
partitions, err := consumer.Partitions(c.Topic)
if err != nil {
log.Fatal("Partitions err: ", err)
}
var wg sync.WaitGroup
wg.Add(len(partitions))
// 然后每个分区开一个 goroutine 来消费
for _, partitionId := range partitions {
//不开异步会导致一个消费完才会消费另外一个
go c.consumeByPartition(consumer, c.Topic, partitionId, &wg)
}
wg.Wait()
}
// 暂时只是业务一对一，也就是一个生产者产生的消息不会触发多个业务的变动
// 但是可以开多个消费者增加处理能力
func getOffsetCacheKey(topic string, partitionId int32) string {
return fmt.Sprintf("kafka_offset_%s_%d", topic, partitionId)
}
func setConsumeOffset(topic string, partitionId int32, offset int64) {
core.RedisBy(core.RedisTypeServer).SetInt64(getOffsetCacheKey(topic, partitionId), offset)
}
func getConsumeOffset(topic string, partitionId int32) (offset int64) {
key := getOffsetCacheKey(topic, partitionId)
if core.RedisBy(core.RedisTypeServer).Exists(key) {
return core.RedisBy(core.RedisTypeServer).GetInt64(key) + 1
}
//默认从最新开始
setConsumeOffset(topic, partitionId, sarama.OffsetNewest)
return sarama.OffsetNewest
}
func (c *KafkaConsumer) consumeByPartition(consumer sarama.Consumer, topic string, partitionId int32, wg *sync.WaitGroup) {
defer wg.Done()
offset := getConsumeOffset(topic, partitionId)
partitionConsumer, err := consumer.ConsumePartition(topic, partitionId, offset)
if err != nil {
log.Fatal("ConsumePartition err: ", err)
}
defer partitionConsumer.Close()
for message := range partitionConsumer.Messages() {
log.Printf("[Consumer] topic: %s ; partitionid: %d; offset:%d, value: %s\n", topic, message.Partition, message.Offset, string(message.Value))
setConsumeOffset(topic, partitionId, message.Offset)
c.MessageQueue <- message.Value
}
}package kafka
import (
"log"
"testing"
"vliao.com/stellar/internal/core"
)
func Test_Get(t *testing.T) {
core.TestMain()
topic := "test_log"
var kafkaConsumer = NewKafkaConsumer(topic)
kafkaConsumer.MessageQueue = make(chan []byte, 1000)
go kafkaConsumer.Consume()
for {
msg := <-kafkaConsumer.MessageQueue
deal(msg)
}
}
func deal(msg []byte) {
log.Printf(string(msg))
}当你停下来休息的时候，不要忘记别人还在奔跑！