前言
消费者主动拉取数据，消息收到后清除消息- 可以有多个topic主题（浏览、点赞、收藏、评论等）。
- 消费者消费数据之后，不删除数据。
- 每个消费者相互独立，都可以消费到数据。
在上篇文章Docker安装kafka、搭建kafka集群中，我们利用docker搭建了kafka集群，在本文中，我们将基于kafka的发布订阅模式，在Golang中使用kafka来实现消息队列。
一、生产者
生产者的实现有两种方式，一种是基于同步消息模式，另一种则是基于异步消息模式。同步消息模式发送完一条消息需要进行确认消息是否到达存储。异步消息模式和同步消息模式的过程大致相似，只不过异步消息生产者不需要在每次发送之后等待接收消息状态（是否成功），下面简单总结下两者的构建步骤。
同步消息模式：
- 构建集群brokers和同步生产者配置config。
- 连接kafka,使用配置构建一个同步生产者。
- 构建发送的消息，每次发送都要重新构建。
- 发送消息，发送后能获取到消息发送的分区和偏移。
异步消息模式：
- 构建集群brokers和异步生产者配置config。
- 连接kafka，使用配置构建一个异步生产者。
- 因为是异步发送，因此需要先启动协程，从不同通道中接收消息状态。
- 构建消息，将消息发送到通道中。
最后，无论是同步生产者还是异步生产者，都别忘了进行资源关闭。
package main
import (
"fmt"
"github.com/Shopify/sarama"
"time"
)
// 基于sarama第三方库开发的kafka client
var brokers = []string{"IP:9092", "IP:9093", "IP:9094"}
var topic = "hello_kafka0"
// 同步消息模式
func syncProducer(config *sarama.Config) {
// 连接kafka,使用配置构建一个同步生产者
syncProducer, err := sarama.NewSyncProducer(brokers, config)
if err != nil {
fmt.Println("syncProducer closed,err:", err)
return
}
defer syncProducer.Close()
//构建发送消息
srcValue := "test syncProducer send msg, i = %d"
for i := 0; i < 5; i++ {
value := fmt.Sprintf(srcValue, i)
msg := &sarama.ProducerMessage{
Topic: topic,
Value: sarama.ByteEncoder(value),
}
// 发送消息，并获取消息存储的分区和偏移
partition, offset, err := syncProducer.SendMessage(msg)
if err != nil {
fmt.Println("send msg failed,err:", err)
return
}
fmt.Printf("send success, partition:%v offset:%v\n", partition, offset)
}
}
// 异步消息模式
func asyncProducer(config *sarama.Config) {
// 连接kafka,使用配置构建一个异步的生产者
asyncProducer, err := sarama.NewAsyncProducer(brokers, config)
if err != nil {
fmt.Println("asyncProducer closed,err:", err)
return
}
defer asyncProducer.AsyncClose() //异步关闭
fmt.Println("start goroutine...")
// 异步发送，因此接收需要先启动协程，从通道中进行接收
go func(producer sarama.AsyncProducer) {
for {
select {
case suc := <-producer.Successes():
fmt.Println("offset: ", suc.Offset, "timestamp:", suc.Timestamp.String(), "partition:", suc.Partition)
case fail := <-producer.Errors():
fmt.Println("err: ", fail.Err)
}
}
}(asyncProducer)
//每500ms构建一条消息进行发送,注意消息每次都需要重新构建
for i := 0; i < 5; i++ {
time.Sleep(500 * time.Millisecond)
timeNow := time.Now()
value := "this is a message " + timeNow.Format("14:49:05")
msg := &sarama.ProducerMessage{ //消息需要每次进行构建
Topic: topic,
Value: sarama.ByteEncoder(value), //将字符串转化为字节数组
}
asyncProducer.Input() <- msg // 使用通道进行发送
}
}
func main() {
config := sarama.NewConfig() //创建一个sarama的config对象
config.Producer.RequiredAcks = sarama.WaitForAll //发送完数据需要isr中的节点，理解为leader和flower都需要回复确认
config.Producer.Partitioner = sarama.NewRandomPartitioner //新选一个patition
//是否等待成功和失败后的响应,只有上面的RequireAcks设置不是NoReponse这里才有用.
config.Producer.Return.Errors = true //接收错误
config.Producer.Return.Successes = true //成功交付的消息将在success channel返回
config.Version = sarama.V3_2_0_0 //指定版本
config.Producer.Retry.Max = 10 //最大重试时间
config.Producer.MaxMessageBytes = 32 * 1024 * 1024 // 最大的消息缓冲字节 默认为100*1024*1024
syncProducer(config)
//asyncProducer(config)
}
二、消费者
消费者构建的一般步骤为：
- 构建集群brokers和消费者配置config。
- 利用配置构建消费者。
- 根据topic主题信息获取该主题存在的所有分区信息。
- 针对每个分区，创建一个分区消费者进行消费，分区消费者接收消息进行消费。
此外，消费者还可以加入给定主题列表的消费者集群，并通过 ConsumerGroupHandler 启动阻塞的 ConsumerGroupSession，下面也给出了实现。
package main
import (
"context"
"fmt"
"github.com/Shopify/sarama"
"sync"
"time"
)
// kafka消费者消费消息
var topic string = "hello_kafka0"
var brokers = []string{"10.227.4.92:9092", "10.227.4.92:9093", "10.227.4.92:9094"}
var topics = []string{"hello_kafka0"}
// 普通消费者
func ordinaryConsumer(wg *sync.WaitGroup, groupId string) {
defer wg.Done() //计数减1
config := sarama.NewConfig()
config.Consumer.Return.Errors = true //是否接收错误
config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange //消费者组的消费策略
config.Consumer.MaxWaitTime = 500 * time.Second //消费者拉取的最大等待时间
config.Version = sarama.V3_2_0_0
config.Consumer.Group.InstanceId = groupId
consumer, err := sarama.NewConsumer(brokers, config)
if err != nil {
fmt.Println("fail to start consumer,err:%v\n", err)
return
}
defer consumer.Close()
partitionList, err := consumer.Partitions(topic) //根据topic获取到所有的分区
if err != nil {
fmt.Printf("fail to get list of partition:err%v\n", err)
return
}
for partition := range partitionList { //遍历所有的分区
//对每个分区创建一个分区消费者,Offset这里指定为获取所有消息，只获取最新的采用OffsetNewest
partConsumer, err := consumer.ConsumePartition(topic, int32(partition), sarama.OffsetOldest)
if err != nil {
fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
return
}
defer partConsumer.AsyncClose()
// 方式1、采用for range方式获取，获取完毕就结束
go func(sarama.PartitionConsumer) {
for msg := range partConsumer.Messages() {
fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v\n",
msg.Partition, msg.Offset, msg.Key, string(msg.Value))
}
}(partConsumer)
time.Sleep(3 * time.Second) //延迟主线程，防止协程还没运行
// 方式2、采用for select方式获取，一直阻塞等待获取
// 信号关闭触发
// signals := make(chan os.Signal, 1)
// signal.Notify(signals, os.Interrupt)
//Loop:
// for {
// select {
// case msg := <-partConsumer.Messages():
// fmt.Printf("Partition:%d Offset:%d Key:%v Value:%v\n",
// msg.Partition, msg.Offset, msg.Key, string(msg.Value))
// case err := <-partConsumer.Errors():
// fmt.Println(err.Err)
// case <-signals:
// break Loop
// }
// }
}
}
// 消费者组,ConsumerGroup负责将主题和分区的处理划分为一组进程（consumer组的成员）
type consumerGroupHandler struct{}
// ConsumerGroupClaim 负责处理来自消费者组中给定主题和分区的Kafka消息
// ConsumerGroupHandler 实例用于处理单个主题/分区声明。 它还为您的消费者组会话生命周期提供钩子，并允许您在消费循环之前或之后触发逻辑。
func (consumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
func (consumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (handler consumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for msg := range claim.Messages() {
fmt.Printf("Message topic:%q partition:%d offset:%d value:%s\n", msg.Topic, msg.Partition, msg.Offset, msg.Value)
sess.MarkMessage(msg, "") //标记这条消息已经消费
}
return nil
}
func groupConsumer(wg *sync.WaitGroup, groupId string) {
defer wg.Done()
config := sarama.NewConfig()
config.Version = sarama.V3_2_0_0
config.Consumer.Return.Errors = true
consumerGroup, err := sarama.NewConsumerGroup(brokers, groupId, config)
if err != nil {
fmt.Println("consumerGroup start failed", err)
return
}
defer func() { _ = consumerGroup.Close() }()
// 启动协程从错误通道中接收错误信息
go func() {
for err := range consumerGroup.Errors() {
fmt.Println("ERROR", err)
}
}()
// 迭代消费者会话
ctx := context.Background()
//`应该在无限循环中调用Consume，当服务器端重新平衡发生时，需要重新创建consumer会话以获取新的声明
for {
handler := consumerGroupHandler{}
err := consumerGroup.Consume(ctx, topics, handler)
if err != nil {
fmt.Println("the Consume failed", err)
return
}
}
}
func main() {
var wg = &sync.WaitGroup{}
wg.Add(2)
//go ordinaryConsumer(wg, "tt")
go groupConsumer(wg, "cc") //通过mark消息已经消费，因此相同消费者组中不会有两个消费者消费到相同的消息
go groupConsumer(wg, "cc")
wg.Wait()
}
三、源码简单解读
sarama.NewConfig()c.Net.MaxOpenRequests = 5
c.Net.DialTimeout = 30 * time.Second
c.Net.ReadTimeout = 30 * time.Second
c.Net.WriteTimeout = 30 * time.Second
c.Net.SASL.Handshake = true
// 元数据配置
c.Metadata.Retry.Max = 3
c.Metadata.Retry.Backoff = 250 * time.Millisecond
c.Metadata.RefreshFrequency = 10 * time.Minute
c.Metadata.Full = true
// 生产者配置
c.Producer.MaxMessageBytes = 1000000 //最大消息字节
c.Producer.RequiredAcks = WaitForLocal //消息确认策略
c.Producer.Timeout = 10 * time.Second //超时时间
c.Producer.Partitioner = NewHashPartitioner //分区器，用于选择主题的分区，策略如下
# sarama.NewManualPartitioner() //返回一个手动选择分区的分割器,也就是获取msg中指定的`partition`
# sarama.NewRandomPartitioner() //通过随机函数随机获取一个分区号
# sarama.NewRoundRobinPartitioner() //环形选择,也就是在所有分区中循环选择一个
# sarama.NewHashPartitioner() //通过msg中的key生成hash值,选择分区,
c.Producer.Retry.Max = 3 //重试次数
c.Producer.Retry.Backoff = 100 * time.Millisecond
c.Producer.Return.Errors = true //是否接收返回的错误消息,当发生错误时会放到Error这个通道中.从它里面获取错误消息
//消费者抓取数据配置
c.Consumer.Fetch.Min = 1
c.Consumer.Fetch.Default = 32768
c.Consumer.Retry.Backoff = 2 * time.Second //失败后再次尝试的间隔时间
c.Consumer.MaxWaitTime = 250 * time.Millisecond //最大等待时间
c.Consumer.MaxProcessingTime = 100 * time.Millisecond
c.Consumer.Return.Errors = false //是否接收返回的错误消息,当发生错误时会放到Error这个通道中.从它里面获取错误消息
c.Consumer.Offsets.CommitInterval = 1 * time.Second // 提交跟新Offset的频率
c.Consumer.Offsets.Initial = OffsetNewest // 指定Offset,也就是从哪里获取消息,默认时从主题的开始获取.
c.ClientID = defaultClientID
c.ChannelBufferSize = 256 //通道缓存大小
c.Version = minVersion //指定kafka版本,不指定,使用最小版本,高版本的新功能可能无法正常使用.
c.MetricRegistry = metrics.NewRegistry()
生产者消息结构ProducerMessage ：
// ProducerMessage is the collection of elements passed to the Producer in order to send a message.
type ProducerMessage struct {
Topic string // 消息主题
// The partitioning key for this message. Pre-existing Encoders include
// StringEncoder and ByteEncoder.
Key Encoder // 消息的分区key的编码方式，这个key用于选择分区,和分割器的NewHashPartitioner联合使用,决定当前消息被保存在哪个分区
// The actual message to store in Kafka. Pre-existing Encoders include
// StringEncoder and ByteEncoder.
Value Encoder //消息的内容
// The headers are key-value pairs that are transparently passed
// by Kafka between producers and consumers.
Headers []RecordHeader //在生产者和消费者之间传递的键值对
// This field is used to hold arbitrary data you wish to include so it
// will be available when receiving on the Successes and Errors channels.
// Sarama completely ignores this field and is only to be used for
// pass-through data.
Metadata interface{} //sarama 用于传递数据使用
// Below this point are filled in by the producer as the message is processed
//Offset、Partition和Timestamp的内容都是由生产者返回后的内容填充.
// Offset is the offset of the message stored on the broker. This is only
// guaranteed to be defined if the message was successfully delivered and
// RequiredAcks is not NoResponse.
Offset int64 //偏移
// Partition is the partition that the message was sent to. This is only
// guaranteed to be defined if the message was successfully delivered.
Partition int32
// Timestamp can vary in behavior depending on broker configuration, being
// in either one of the CreateTime or LogAppendTime modes (default CreateTime),
// and requiring version at least 0.10.0.
//
// When configured to CreateTime, the timestamp is specified by the producer
// either by explicitly setting this field, or when the message is added
// to a produce set.
//
// When configured to LogAppendTime, the timestamp assigned to the message
// by the broker. This is only guaranteed to be defined if the message was
// successfully delivered and RequiredAcks is not NoResponse.
Timestamp time.Time
retries int // 重试次数
flags flagSet
expectation chan *ProducerError
sequenceNumber int32
producerEpoch int16
hasSequence bool
}
消费者消息结构ConsumerMessage：
// ConsumerMessage encapsulates a Kafka message returned by the consumer.
type ConsumerMessage struct {
Headers []*RecordHeader // only set if kafka is version 0.11+
Timestamp time.Time // only set if kafka is version 0.10+, inner message timestamp
BlockTimestamp time.Time // only set if kafka is version 0.10+, outer (compressed) block timestamp
Key, Value []byte //key和保存的值
Topic string //要消费的主题
Partition int32 //要消费的分区
Offset int64 //要消费的消息的位置,从哪里开始消费,最开始的,还是最后的
}