Golang中如何正確的使用sarama包操作Kafka？
一、背景
在一些業務系統中，模塊之間通過引入Kafka解藕，拿IM舉例（圖來源）：
用戶A給B發送訊息，msg_gateway收到訊息后，投遞訊息到Kafka后就給A回傳發送成功，這個時候，其實還沒有持久化到mysql中，雖然最侄訓保持一致性，所以，試想如果Kafka丟訊息了，是不是就出大問題了？A認為給B發送訊息成功了，但是在服務器內部訊息丟失了B并沒有收到，
所以，在使用Kafka的時候，有一些業務對訊息丟失問題非常的關注，
同樣，常見的問題還有：
- 重復消費的問題，
- 亂序的問題，
二、Kafka訊息丟失問題描述
以下內容來源 kafka什么時候會丟訊息 上面我們擔心的點需要進一步明確一下丟訊息的定義：kafka集群中的部分或全部broker掛了，導致consumer沒有及時收到訊息，這不屬于丟訊息，broker掛了，只要訊息全部持久化到了硬碟上，重啟broker集群之后，使消費者繼續拉取訊息，訊息就沒有丟失，仍然全量消費了，所以我的理解，所謂丟訊息，意味著：開發人員未感知到哪些訊息沒有被消費， 作者把訊息的丟失歸納了以下幾種情況： 1） producer把訊息發送給broker，因為網路抖動，訊息沒有到達broker，且開發人員無感知， 解決方案：producer設定acks引數，訊息同步到master之后回傳ack信號，否則拋例外使應用程式感知到并在業務中進行重試發送，這種方式一定程度保證了訊息的可靠性，producer等待broker確認信號的時延也不高， 2）producer把訊息發送給broker-master，master接收到訊息，在未將訊息同步給follower之前，掛掉了，且開發人員無感知， 解決方案：producer設定acks引數，訊息同步到master且同步到所有follower之后回傳ack信號，否則拋例外使應用程式感知到并在業務中進行重試發送，這樣設定，在更大程度上保證了訊息的可靠性，缺點是producer等待broker確認信號的時延比較高， 3）producer把訊息發送給broker-master，master接收到訊息，master未成功將訊息同步給每個follower，有訊息丟失風險， 解決方案：同上， 4）某個broker訊息尚未從記憶體緩沖區持久化到磁盤，就掛掉了，這種情況無法通過ack機制感知， 解決方案：設定引數，加快訊息持久化的頻率，能在一定程度上減少這種情況發生的概率，但提高頻率自然也會影響性能， 5）consumer成功拉取到了訊息，consumer掛了， 解決方案：設定手動sync，消費成功才提交， 綜上所述，集群/專案運轉正常的情況下，kafka不會丟訊息，一旦集群出現問題，訊息的可靠性無法完全保證，要想盡可能保證訊息可靠，基本只能在發現訊息有可能沒有被消費時，重發訊息來解決，所以在業務邏輯中，要考慮訊息的重復消費問題，對于關鍵環節，要有冪等機制， 作者的幾條建議： 1）如果一個業務很關鍵，使用kafka的時候要考慮丟訊息的成本和解決方案， 2）producer端確認訊息是否到達集群，若有例外，進行重發， 3）consumer端保障消費冪等性， 4）運維保障集群運轉正常且高可用，保障網路狀況良好，三、生產端丟訊息問題解決
上面說了，只需要把producer設定acks引數，等待Kafka所有follower都成功后再回傳，我們只需要進行如下設定：config := sarama.NewConfig()
config.Producer.RequiredAcks = sarama.WaitForAll // -1ack引數有如下取值：
const (
// NoResponse doesn't send any response, the TCP ACK is all you get.
NoResponse RequiredAcks = 0
// WaitForLocal waits for only the local commit to succeed before responding.
WaitForLocal RequiredAcks = 1
// WaitForAll waits for all in-sync replicas to commit before responding.
// The minimum number of in-sync replicas is configured on the broker via
// the `min.insync.replicas` configuration key.
WaitForAll RequiredAcks = -1
)
四、消費端丟訊息問題
通常消費端丟訊息都是因為Offset自動提交了，但是資料并沒有插入到mysql（比如出現BUG或者行程Crash），導致下一次消費者重啟后，訊息漏掉了，自然資料庫中也查不到，這個時候，我們可以通過手動提交解決，甚至在一些復雜場景下，還要使用二階段提交，自動提交模式下的丟訊息問題
默認情況下，sarama是自動提交的方式，間隔為1秒鐘// NewConfig returns a new configuration instance with sane defaults.
func NewConfig() *Config {
// …
c.Consumer.Offsets.AutoCommit.Enable = true. // 自動提交
c.Consumer.Offsets.AutoCommit.Interval = 1 * time.Second // 間隔
c.Consumer.Offsets.Initial = OffsetNewest
c.Consumer.Offsets.Retry.Max = 3
// ...
}這里的自動提交，是基于被標記過的訊息（sess.MarkMessage(msg, “")）
type exampleConsumerGroupHandler struct{}
func (exampleConsumerGroupHandler) Setup(_ ConsumerGroupSession) error { return nil }
func (exampleConsumerGroupHandler) Cleanup(_ ConsumerGroupSession) error { return nil }
func (h exampleConsumerGroupHandler) ConsumeClaim(sess ConsumerGroupSession, claim ConsumerGroupClaim) error {
for msg := range claim.Messages() {
fmt.Printf("Message topic:%q partition:%d offset:%d\n", msg.Topic, msg.Partition, msg.Offset)
// 標記訊息已處理，sarama會自動提交
sess.MarkMessage(msg, "")
}
return nil
}如果不呼叫sess.MarkMessage(msg, “")，即使啟用了自動提交也沒有效果，下次啟動消費者會從上一次的Offset重新消費，我們不妨注釋掉sess.MarkMessage(msg, “")，然后打開Offset Explorer查看：
那么這樣，我們大概理解了sarama自動提交的原理：先標記再提交，我們只需要保持標記邏輯在插入mysql代碼之后即可確保不會出現丟訊息的問題：
正確的呼叫順序：
func (h msgConsumerGroup) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for msg := range claim.Messages() {
// 插入mysql
insertToMysql(msg)
// 正確：插入mysql成功后程式崩潰，下一次頂多重復消費一次，而不是因為Offset超前，導致應用層訊息丟失了
sess.MarkMessage(msg, “")
}
return nil
}錯誤的順序：
func (h msgConsumerGroup) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for msg := range claim.Messages() {
// 錯誤1：不能先標記，再插入mysql，可能標記的時候剛好自動提交Offset，但mysql插入失敗了，導致下一次這個訊息不會被消費，造成丟失
// 錯誤2：干脆忘記呼叫sess.MarkMessage(msg, “")，導致重復消費
sess.MarkMessage(msg, “")
// 插入mysql
insertToMysql(msg)
}
return nil
}sarama手動提交模式
當然，另外也可以通過手動提交來處理丟訊息的問題，但是個人不推薦，因為自動提交模式下已經能解決丟訊息問題，consumerConfig := sarama.NewConfig()
consumerConfig.Version = sarama.V2_8_0_0
consumerConfig.Consumer.Return.Errors = false
consumerConfig.Consumer.Offsets.AutoCommit.Enable = false // 禁用自動提交，改為手動
consumerConfig.Consumer.Offsets.Initial = sarama.OffsetNewest
func (h msgConsumerGroup) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for msg := range claim.Messages() {
fmt.Printf("%s Message topic:%q partition:%d offset:%d value:%s\n", h.name, msg.Topic, msg.Partition, msg.Offset, string(msg.Value))
// 插入mysql
insertToMysql(msg)
// 手動提交模式下，也需要先進行標記
sess.MarkMessage(msg, "")
consumerCount++
if consumerCount%3 == 0 {
// 手動提交，不能頻繁呼叫，耗時9ms左右，macOS i7 16GB
t1 := time.Now().Nanosecond()
sess.Commit()
t2 := time.Now().Nanosecond()
fmt.Println("commit cost:", (t2-t1)/(1000*1000), "ms")
}
}
return nil
}
五、Kafka訊息順序問題
投遞Kafka之前，我們通過一次gRPC呼叫解決了訊息序號的生成問題，但是這里其實還涉及一個訊息順序問題：訂閱Kafka的消費者如何按照訊息順序寫入mysql，而不是隨機寫入呢？ 我們知道，Kafka的訊息在一個partition中是有序的，所以只要確保發給某個人的訊息都在同一個partition中即可， 1.全域一個partition 這個最簡單，但是在kafka中一個partition對應一個執行緒，所以這種模型下Kafka的吞吐是個問題， 2.多個partition，手動指定msg := &sarama.ProducerMessage{
Topic: “msgc2s",
Value: sarama.StringEncoder(“hello”),
Partition: toUserId % 10,
}
partition, offset, err := producer.SendMessage(msg)生產訊息的時候，除了Topic和Value，我們可以通過手動指定partition，比如總共有10個磁區，我們根據用戶ID取余，這樣發給同一個用戶的訊息，每次都到1個partition里面去了，消費者寫入mysql中的時候，自然也是有序的，
但是，因為磁區總數是寫死的，萬一Kafka的磁區數要調整呢？那不得重新編譯代碼？所以這個方式不夠優美，
3.多個partition，自動計算
kafka客戶端為我們提供了這種支持，首先，在初始化的時候，設定選擇磁區的策略為Hash
p.config.Producer.Partitioner = sarama.NewHashPartitioner然后，在生成訊息之前，設定訊息的Key值：
msg := &sarama.ProducerMessage{
Topic: "testAutoSyncOffset",
Value: sarama.StringEncoder("hello"),
Key: sarama.StringEncoder(strconv.Itoa(RecvID)),
}4.擴展知識：多執行緒情況下一個partition的亂序處理
我們上面說了，Kafka客戶端針對一個partition開一個執行緒進行消費，如果處理比較耗時的話，比如處理一條訊息耗時幾十 ms，那么 1 秒鐘就只能處理幾十條訊息，這吞吐量太低了，這個時候，我們可能就把邏輯移動到其他執行緒里面去處理，這樣的話，順序就可能會亂，
我們可以通過寫 N 個記憶體 queue，具有相同 key 的資料都到同一個記憶體 queue；然后對于 N 個執行緒，每個執行緒分別消費一個記憶體 queue 即可，這樣就能保證順序性，PS：就像4 % 10 = 4，14 % 10 = 4，他們取余都是等于4，所以落到了一個partition，但是key值不一樣啊，我們可以自己再取余，放到不同的queue里面，
六、重復消費和訊息冪等
這篇文章中：kafka什么時候會丟訊息 詳細了描述了各種丟訊息的情況，我們通過設定 RequiredAcks = sarama.WaitForAll（-1），可以解決生產端丟訊息的問題，第六節中也對消費端丟訊息進行了說明，只需要確保在插入資料庫之后，呼叫sess.MarkMessage(msg, "”)即可， 如果出現了插入Mysql成功，但是因為自動提交有1秒的間隔，如果此時崩潰，下次啟動消費者勢必會對者1秒的資料進行重復消費，我們在應用層需要處理這個問題， 常見的有2種思路：- 如果是存在redis中不需要持久化的資料，比如string型別，set具有天然的冪等性，無需處理，
- 插入mysql之前，進行一次query操作，針對每個客戶端發的訊息，我們為它生成一個唯一的ID（比如GUID），或者直接把訊息的ID設定為唯一索引，
七、完整代碼實體
consumer.gotype msgConsumerGroup struct{}
func (msgConsumerGroup) Setup(_ sarama.ConsumerGroupSession) error { return nil }
func (msgConsumerGroup) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (h msgConsumerGroup) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for msg := range claim.Messages() {
fmt.Printf("%s Message topic:%q partition:%d offset:%d value:%s\n", h.name, msg.Topic, msg.Partition, msg.Offset, string(msg.Value))
// 查mysql去重
if check(msg) {
// 插入mysql
insertToMysql()
}
// 標記，sarama會自動進行提交，默認間隔1秒
sess.MarkMessage(msg, "")
}
return nil
}
func main(){
consumerConfig := sarama.NewConfig()
consumerConfig.Version = sarama.V2_8_0_0 // specify appropriate version
consumerConfig.Consumer.Return.Errors = false
//consumerConfig.Consumer.Offsets.AutoCommit.Enable = true // 禁用自動提交，改為手動
//consumerConfig.Consumer.Offsets.AutoCommit.Interval = time.Second * 1 // 測驗3秒自動提交
consumerConfig.Consumer.Offsets.Initial = sarama.OffsetNewest
cGroup, err := sarama.NewConsumerGroup([]string{"10.0.56.153:9092", "10.0.56.153:9093", "10.0.56.153:9094"},"testgroup", consumerConfig)
if err != nil {
panic(err)
}
for {
err := cGroup.Consume(context.Background(), []string{"testAutoSyncOffset"}, consumerGroup)
if err != nil {
fmt.Println(err.Error())
break
}
}
_ = cGroup.Close()
}producer.go
func main(){
config := sarama.NewConfig()
config.Producer.RequiredAcks = sarama.WaitForAll // 等待所有follower都回復ack，確保Kafka不會丟訊息
config.Producer.Return.Successes = true
config.Producer.Partitioner = sarama.NewHashPartitioner // 對Key進行Hash，同樣的Key每次都落到一個磁區，這樣訊息是有序的
// 使用同步producer，異步模式下有更高的性能，但是處理更復雜，這里建議先從簡單的入手
producer, err := sarama.NewSyncProducer([]string{"10.0.56.153:9092"}, config)
defer func() {
_ = producer.Close()
}()
if err != nil {
panic(err.Error())
}
msgCount := 4
// 模擬4個訊息
for i := 0; i < msgCount; i++ {
rand.Seed(int64(time.Now().Nanosecond()))
msg := &sarama.ProducerMessage{
Topic: "testAutoSyncOffset",
Value: sarama.StringEncoder("hello+" + strconv.Itoa(rand.Int())),
Key: sarama.StringEncoder("BBB”),
}
t1 := time.Now().Nanosecond()
partition, offset, err := producer.SendMessage(msg)
t2 := time.Now().Nanosecond()
if err == nil {
fmt.Println("produce success, partition:", partition, ",offset:", offset, ",cost:", (t2-t1)/(1000*1000), " ms")
} else {
fmt.Println(err.Error())
}
}
}
八、參考
- Kafka 的資料丟失和重復消費 https://zhuanlan.zhihu.com/p/54287819
- kafka什么時候會丟訊息
- CAP 定理的含義 https://www.ruanyifeng.com/blog/2018/07/cap.html
- Kafka入門（3）：Sarama生產者是如何作業的 https://www.cnblogs.com/hongjijun/p/13584373.html
- 超好用的 Kafka 客戶端管理工具 Offset Explorer http://www.ibloger.net/article/3497.html
- 查看集群中kafka的Version（版本） https://blog.csdn.net/Damonhaus/article/details/54310868
- Kafka如何保證訊息的順序性 https://blog.csdn.net/qianshangding0708/article/details/103360193
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/295962.html
標籤：Go