原文链接：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 = 0WaitForLocal RequiredAcks = 1WaitForAll 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...