下面是一個包含 3 個 Go 例程的服務(wù)，用于處理來自 Kafka 的消息：通道 1 和通道 2 是 Go 中的無緩沖數(shù)據(jù)通道。通道就像一個排隊機(jī)制。Goroutine-1 從 kafka 主題讀取消息，在驗證消息后將其消息負(fù)載拋出到通道-1 上。Goroutine-2 從通道 1 讀取并處理有效負(fù)載，并將處理后的有效負(fù)載拋出通道 2。Goroutine-3 從通道 2 讀取數(shù)據(jù)，并將處理后的有效負(fù)載封裝到 http 數(shù)據(jù)包中，并向另一個服務(wù)執(zhí)行 http 請求（使用 http 客戶端）。上述流程中的漏洞：在我們的例子中，由于服務(wù)之間的網(wǎng)絡(luò)連接不良或遠(yuǎn)程服務(wù)未準(zhǔn)備好接受來自Go-routine3的http請求（http客戶端超時），處理失敗，因此，上述服務(wù)會丟失該消息（已從Kafka主題中讀?。８牯斖?1目前訂閱來自卡夫卡的消息，而沒有向卡夫卡發(fā)送確認(rèn)（通知戈魯廷-3已成功處理特定消息）正確性優(yōu)先于性能。如何確保每條消息都得到成功處理？
2 回答
白豬掌柜的
TA貢獻(xiàn)1893條經(jīng)驗 獲得超10個贊
例如，通過新的通道-3將戈魯廷-3的反饋添加到戈魯廷-1。戈魯廷-1將阻塞，直到它得到頻道-3的確認(rèn)。
// in gorouting 1
channel1 <- data
select {
case <-channel3:
case <-ctx.Done(): // or smth else to prevent deadlock
}
...
// in gorouting 3
data := <-channel2
for {
if err := sendData(data); err == nil {
break
}
}
channel3<-struct{}{}
撒科打諢
TA貢獻(xiàn)1934條經(jīng)驗 獲得超2個贊
為了確保正確性，您需要在處理成功完成后提交（=確認(rèn)）消息。
對于處理未成功完成的情況 - 通常，您需要自己實現(xiàn)重試機(jī)制。
這應(yīng)該特定于您的用例，但通常您將消息拋回專用的Kafka重試主題（您創(chuàng)建），添加睡眠并再次處理消息。如果在 x 次后處理失敗 - 則將消息拋出到 DLQ（=死信隊列）。
你可以在這里閱讀更多：
https://eng.uber.com/reliable-reprocessing/
https://www.confluent.io/blog/error-handling-patterns-in-kafka/
- 2 回答
- 0 關(guān)注
- 102 瀏覽
添加回答
舉報
0/150
提交
取消