关于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 信息。

先看一下 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 机器都会做备份，这种状况很极其，能够忽略不计。

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

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

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

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

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

次要是通过两点来解决：

通过设置 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
}

这个解决办法就比拟粗犷了，间接应用主动提交的模式，在每次真正生产完之后在本人手动提交 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…

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

背景

初识 `kafka` 架构

kafka 丢音讯的三个节点

生产者 push 音讯节点

kafka 集群本身故障造成

消费者 pull 音讯节点

实战

解决 push 音讯失落问题

解决 pull 音讯失落问题

总结

参考文章

背景

初识 kafka 架构

kafka 丢音讯的三个节点

生产者 push 音讯节点

kafka 集群本身故障造成

消费者 pull 音讯节点

实战

解决 push 音讯失落问题

解决 pull 音讯失落问题

总结

参考文章

初识 `kafka` 架构