无消息丢失配置怎么实现?
-
Kafka 只对“已提交”的消息(committed message)做有限度的持久化保证。
- 第一个核心要素是“已提交的消息”。
- 当 Kafka 的若干个 Broker 成功地接收到一条消息并写入到日志文件后,它们会告诉生产者程序这条消息已成功提交。
- 可以选择只要有一个 Broker 成功保存该消息就算是已提交,也可以是令所有 Broker 都成功保存该消息才算是已提交。
- 第二个核心要素就是“有限度的持久化保证”。
- Kafka 不可能保证在任何情况下都做到不丢失消息。
- Kafka 不丢消息是有前提条件的。假如你的消息保存在 N 个 Kafka Broker 上,那么这个前提条件就是这 N 个 Broker 中至少有 1 个存活。
- 第一个核心要素是“已提交的消息”。
消息丢失案例
案例 1:生产者程序丢失数据
- 目前 Kafka Producer 是异步发送消息的,也就是说如果你调用的是 producer.send(msg) 这个 API,那么它通常会立即返回,但此时你不能认为消息发送已成功完成。
- 如果用这个方式,可能会有哪些因素导致消息没有发送成功呢?
- 网络抖动,导致消息压根就没有发送到 Broker 端;
- 消息本身不合格导致 Broker 拒绝接收(比如消息太大了,超过了 Broker 的承受能力)。
- 解决此问题的方法非常简单:
- Producer 永远要使用带有回调通知的发送 API,也就是说不要使用 producer.send(msg),而要使用 producer.send(msg, callback)。
案例 2:消费者程序丢失数据
- Consumer 端丢失数据主要体现在 Consumer 端要消费的消息不见了。
- Consumer 程序有个“位移”的概念,表示的是这个 Consumer 当前消费到的 Topic 分区的位置。
- 只要维持先消费消息,再更新位移的顺序即可。这样就能最大限度地保证消息不丢失。这种处理方式可能带来的问题是消息的重复处理。
- Consumer 程序有个“位移”的概念,表示的是这个 Consumer 当前消费到的 Topic 分区的位置。
- 还存在一种比较隐蔽的消息丢失场景。
- Consumer 程序从 Kafka 获取到消息后开启了多个线程异步处理消息,而 Consumer 程序自动地向前更新位移。
- 假如其中某个线程运行失败了,它负责的消息没有被成功处理,但位移已经被更新了,因此这条消息对于 Consumer 而言实际上是丢失了。
- 这里的关键在于 Consumer 自动提交位移,你没有真正地确认消息是否真的被消费就“盲目”地更新了位移。
- 这个问题的解决方案也很简单:如果是多线程异步处理消费消息,Consumer 程序不要开启自动提交位移,而是要应用程序手动提交位移。
最佳实践
- 不要使用 producer.send(msg),而要使用 producer.send(msg, callback)。记住,一定要使用带有回调通知的 send 方法。
-
设置 acks = all。
- acks 是 Producer 的一个参数,代表了你对“已提交”消息的定义。
- 如果设置成 all,则表明所有副本 Broker 都要接收到消息,该消息才算是“已提交”。
- 这是最高等级的“已提交”定义。
-
设置 retries 为一个较大的值。
- 这⾥的 retries 同样是 Producer 的参数,对应 Producer 自动重试。
- 当出现网络的瞬时抖动时,消息发送可能会失败,此时配置了 retries > 0 的 Producer 能够自动重试消息发送,避免消息丢失。
-
设置 unclean.leader.election.enable = false。
- 这是 Broker 端的参数,它控制的是哪些 Broker 有资格竞选分区的 Leader。
- 如果一个 Broker 落后原先的 Leader 太多,那么它一旦成为新的 Leader,必然会造成消息的丢失。
- 故一般都要将该参数设置成 false,即不允许这种情况的发生。
-
设置 replication.factor >= 3。
- 这也是 Broker 端的参数。
- 其实这里想表述的是,最好将消息多保存几份,毕竟目前防止消息丢失的主要机制就是冗余。
-
设置 min.insync.replicas > 1。
- 这依然是 Broker 端参数,控制的是消息至少要被写入到多少个副本才算是“已提交”。
- 设置成大于 1 可以提升消息持久性。
- 在实际环境中千万不要使用默认值 1。
-
确保 replication.factor > min.insync.replicas。
- 如果两者相等,那么只要有一个副本挂机,整个分区就无法正常工作了。
- 我们不仅要改善消息的持久性,防止数据丢失,还要在不降低可用性的基础上完成。
- 推荐设置成 replication.factor = min.insync.replicas + 1。
-
确保消息消费完成再提交。
- Consumer 端有个参数 enable.auto.commit,最好把它设置成 false,并采用手动提交位移的方式。
文章来源地址https://www.toymoban.com/news/detail-602083.html
文章来源:https://www.toymoban.com/news/detail-602083.html
到了这里,关于《kafka 核心技术与实战》课程学习笔记(八)的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!