关于golang:Go语言如何操纵Kafka保证无消息丢失

原文链接：Go语言如何操纵Kafka保障无音讯失落

背景

kafkakafkaGoKafka

kafka

kafka

kafka

Kafka是由Apache软件基金会开发的一个开源流解决平台，由Scala和Java编写。该项目标指标是为解决实时数据提供一个对立、高吞吐、低提早的平台。其长久化层实质上是一个“依照分布式事务日志架构的大规模公布/订阅音讯队列”，这使它作为企业级基础设施来解决流式数据十分有价值。此外，Kafka能够通过Kafka Connect连贯到内部零碎（用于数据输出/输入），并提供了Kafka Streams——一个Java]流式解决库。
该设计受事务日志的影响较大。

producerbrokerconsumer

针对架构图咱们解释一个各个模块：

topictopic

还有些概念咱们也介绍一下：

leaderfowllerleader

kafka丢音讯的三个节点

生产者push音讯节点

先看一下producer的大略写入流程：

• producer先从kafka集群找到该partition的leader
• producer将音讯发送给leader，leader将该音讯写入本地
• follwers从leader pull音讯，写入本地log后leader发送ack
• leader 收到所有 ISR 中的 replica 的 ACK 后，减少high watermark，并向 producer 发送 ack

通过这个流程咱们能够看到kafka最终会返回一个ack来确认推送音讯后果，这里kafka提供了三种模式：

NoResponse RequiredAcks = 0
WaitForLocal RequiredAcks = 1
WaitForAll RequiredAcks = -1

NoResponse RequiredAcks = 0WaitForLocal RequiredAcks = 1WaitForAll RequiredAcks = -1

所以依据这三种模式咱们就能推断出生产者在push音讯时有肯定几率失落的，剖析如下：

123

所以在生产环境中咱们能够抉择模式2或者模式3来保障音讯的可靠性，具体须要依据业务场景来进行抉择，在乎吞吐量就抉择模式2，不在乎吞吐量，就抉择模式3，要想齐全保证数据不失落就抉择模式3是最牢靠的。

kafka集群本身故障造成

kafka

消费者pull音讯节点

push音讯时会把数据追加到Partition并且调配一个偏移量，这个偏移量代表以后消费者生产到的地位，通过这个Partition也能够保障音讯的程序性，消费者在pull到某个音讯后，能够设置主动提交或者手动提交commit，提交commit胜利，offset就会产生偏移:

所以主动提交会带来数据失落的问题，手动提交会带来数据反复的问题，剖析如下：

• 在设置主动提交的时候，当咱们拉取到一个音讯后，此时offset曾经提交了，然而咱们在解决生产逻辑的时候失败了，这就会导致数据失落了
• 在设置手动提交时，如果咱们是在解决完音讯后提交commit，那么在commit这一步产生了失败，就会导致反复生产的问题。

比起数据失落，反复生产是合乎业务预期的，咱们能够通过一些幂等性设计来躲避这个问题。

实战

残缺代码曾经上传github：https://github.com/asong2020/…

解决push音讯失落问题

次要是通过两点来解决：

RequiredAcksWaitForAll

因而咱们写出如下代码（摘出创立client局部）：

func NewAsyncProducer() sarama.AsyncProducer {
    cfg := sarama.NewConfig()
    version, err := sarama.ParseKafkaVersion(VERSION)
    if err != nil{
        log.Fatal("NewAsyncProducer Parse kafka version failed", err.Error())
        return nil
    }
    cfg.Version = version
    cfg.Producer.RequiredAcks = sarama.WaitForAll // 三种模式任君抉择
    cfg.Producer.Partitioner = sarama.NewHashPartitioner
    cfg.Producer.Return.Successes = true
    cfg.Producer.Return.Errors = true
    cfg.Producer.Retry.Max = 3 // 设置重试3次
    cfg.Producer.Retry.Backoff = 100 * time.Millisecond
    cli, err := sarama.NewAsyncProducer([]string{ADDR}, cfg)
    if err != nil{
        log.Fatal("NewAsyncProducer failed", err.Error())
        return nil
    }
    return cli
}

解决pull音讯失落问题

这个解决办法就比拟粗犷了，间接应用主动提交的模式，在每次真正生产完之后在本人手动提交offset，然而会产生反复生产的问题，不过很好解决，应用幂等性操作即可解决。

代码示例：

func NewConsumerGroup(group string) sarama.ConsumerGroup {
    cfg := sarama.NewConfig()
    version, err := sarama.ParseKafkaVersion(VERSION)
    if err != nil{
        log.Fatal("NewConsumerGroup Parse kafka version failed", err.Error())
        return nil
    }

    cfg.Version = version
    cfg.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange
    cfg.Consumer.Offsets.Initial = sarama.OffsetOldest
    cfg.Consumer.Offsets.Retry.Max = 3
    cfg.Consumer.Offsets.AutoCommit.Enable = true // 开启主动提交，须要手动调用MarkMessage才无效
    cfg.Consumer.Offsets.AutoCommit.Interval = 1 * time.Second // 距离
    client, err := sarama.NewConsumerGroup([]string{ADDR}, group, cfg)
    if err != nil {
        log.Fatal("NewConsumerGroup failed", err.Error())
    }
    return client
}

下面次要是创立ConsumerGroup局部，仔细的读者应该看到了，咱们这里应用的是主动提交，说好的应用手动提交呢？这是因为咱们这个kafka库的个性不同，这个主动提交须要与MarkMessage()办法配合应用才会提交(有疑难的敌人能够实际一下，或者看一下源码)，否则也会提交失败，因为咱们在写生产逻辑时要这样写：

func (e EventHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    for msg := range claim.Messages() {
        var data common.KafkaMsg
        if err := json.Unmarshal(msg.Value, &data); err != nil {
            return errors.New("failed to unmarshal message err is " + err.Error())
        }
        // 操作数据，改用打印
        log.Print("consumerClaim data is ")

        // 解决音讯胜利后标记为解决, 而后会主动提交
        session.MarkMessage(msg,"")
    }
    return nil
}

或者间接应用手动提交办法来解决，只需两步：

第一步：敞开主动提交：

consumerConfig.Consumer.Offsets.AutoCommit.Enable = false  // 禁用主动提交，改为手动

第二步：生产逻辑中增加如下代码，手动提交模式下，也须要先进行标记，在进行commit

session.MarkMessage(msg,"")
session.Commit()

残缺代码能够到github上下载并进行验证！

总结

本文咱们次要阐明了两个知识点：

• Kafka会产生音讯失落
• 应用Go操作Kafka如何配置能够不失落数据

日常业务开发中，很多公司都喜爱拿音讯队列进行解耦，那么你就要留神了，应用Kafka做音讯队列无奈保证数据不失落，须要咱们本人手动配置弥补，别忘记了，要不又是一场P0事变。

欢送关注公众号：Golang梦工厂

举荐往期文章：

• 学习channel设计：从入门到放弃
• 详解内存对齐
• [警觉] 请勿滥用goroutine
• 源码分析panic与recover，看不懂你打我好了！
• 面试官：小松子来聊一聊内存逃逸
• 面试官：两个nil比拟后果是什么？
• 并发编程包之 errgroup

参考文章

• https://juejin.cn/post/684490…
• https://cloud.tencent.com/dev…
• https://juejin.cn/post/699926…