乐趣区

关于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的整体架构比较简单,次要由 producerbrokerconsumer 组成:

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

  • 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 下的不同分区信息是不同的,同一分区信息是有序的;每一个分区都有一个或者多个正本,其中会选举一个 leaderfowllerleader拉取数据更新本人的 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…
退出移动版