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

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

背景

目前一些互联网公司会应用音讯队列来做外围业务，因为是外围业务，所以对数据的最初一致性比拟敏感，如果两头呈现数据失落，就会引来用户的投诉，年底绩效就变成 325 了。之前和几个敌人聊天，他们的公司都在用 kafka 来做音讯队列，应用 kafka 到底会不会丢音讯呢？如果丢音讯了该怎么做好弥补措施呢？本文咱们就一起来剖析一下，并介绍如何应用 Go 操作 Kafka 能够不失落数据。

本文操作 kafka 基于：https://github.com/Shopify/sa…

初识 kafka 架构

维基百科对 kafka 的介绍：

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

kafka的整体架构比较简单，次要由 producer、broker、consumer 组成：

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

• Producer：数据的生产者，能够将数据公布到所抉择的 topic 中。
• Consumer：数据的消费者，应用 Consumer Group 进行标识，在 topic 中的每条记录都会被调配给订阅生产组中的一个消费者实例，消费者实例能够散布在多个过程中或者多个机器上。
• Broker：消息中间件解决节点（服务器），一个节点就是一个 broker，一个 Kafka 集群由一个或多个 broker 组成。

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

• topic：能够了解为一个音讯的汇合，topic 存储在 broker 中，一个 topic 能够有多个 partition 分区，一个 topic 能够有多个 Producer 来 push 音讯，一个 topic 能够有多个消费者向其 pull 音讯，一个 topic 能够存在一个或多个 broker 中。
• partition：其是 topic 的子集，不同分区调配在不同的 broker 上进行程度扩大从而减少 kafka 并行处理能力，同 topic 下的不同分区信息是不同的，同一分区信息是有序的；每一个分区都有一个或者多个正本，其中会选举一个 leader，fowller 从leader拉取数据更新本人的 log（每个分区逻辑上对应一个 log 文件夹），消费者向 leader 中 pull 信息。

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 = 0：这个代表的就是数据推出的胜利与否都与我无关了
• WaitForLocal RequiredAcks = 1：当 local(leader)确认接管胜利后，就能够返回了
• WaitForAll RequiredAcks = -1：当所有的 leader 和 follower 都接管胜利时，才会返回

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

• 如果咱们抉择了模式1，这种模式失落数据的几率很大，无奈重试
• 如果咱们抉择了模式2，这种模式下只有 leader 不挂，就能够保证数据不失落，然而如果 leader 挂了，follower 还没有同步数据，那么就会有肯定几率造成数据失落
• 如果抉择了模式3，这种状况不会造成数据失落，然而有可能会造成数据反复，如果 leader 与 follower 同步数据是网络呈现问题，就有可能造成数据反复的问题。

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

kafka 集群本身故障造成

kafka 集群接管到数据后会将数据进行长久化存储，最终数据会被写入到磁盘中，在写入磁盘这一步也是有可能会造成数据损失的，因为写入磁盘的时候操作系统会先将数据写入缓存，操作系统将缓存中数据写入磁盘的工夫是不确定的，所以在这种状况下，如果 kafka 机器忽然宕机了，也会造成数据损失，不过这种概率产生很小，个别公司外部 kafka 机器都会做备份，这种状况很极其，能够忽略不计。

消费者 pull 音讯节点

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

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

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

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

实战

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

解决 push 音讯失落问题

次要是通过两点来解决：

• 通过设置 RequiredAcks 模式来解决，选用 WaitForAll 能够保证数据推送胜利，不过会影响时延时
• 引入重试机制，设置重试次数和重试距离

因而咱们写出如下代码（摘出创立 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…