前言
回答这个问题，我想可以从消息流开始产生（生产端），到kafka服务存储（broker），到被消费（消费端）的这个过程着手分析，明确kafka消息丢失的边界和如何配置kafka才能保证消息不丢失。
每个服务都应该有自己明确的职责范围，kafka服务也不例外。所以，kafka保证消息不丢失，首先有一个重要的前提，那就是「消息已提交」。如果一条消息还未到达kafka服务，或者是到达了却未满足「已提交」这个定义，那么在kafka看来都属于消息未提交。
接下来，我们一起来剖析这个过程吧～
生产端
kafka生产者发送消息之后被抱怨丢失消息是我们常常听到的。实际上，kafka生产者提供的send()接口，有三种模式：
- 发后即忘（fire-and-forget）。send(record)接口的调用让生产者只管发送消息而不用关心消息是否正确到达kafka服务。这种发送方式的性能最高，但可靠性也最差，可能会出现消息丢失。
- 同步（sync）。send(record).get()接口的调用会让生产者进程阻塞等待kafka服务的响应。实际上send(record)方法本身是异步的，只是直接链接调用get()方法是同步阻塞的。这种发送方式的可靠性最高，但性能也是差很多，下一条消息的发送需阻塞等待上一条消息发送完之后才能进行。
- 异步（async）。send(record, callback)接口的调用让生产者即不会像第1模式那样完全不在乎是否发送成功，也不像第2模式那样发送需阻塞等待。有了callback回调，消息发送既可以异步，同时如果kafka服务有错误信息，也能准确回调告诉程序，让其能针对性地进行处理。并且，对于同一个分区而言，回调函数的调用也可以保证分区有序。
生产环境应用中，大部分人都会选择第3种模式，让callback回调告诉本次消息是否真的提交成功。另外可能还会因为网络抖动而发送失败的情况，我们可以配置重试机制。总之，生产端要保证消息「已提交」到kafka服务这个职责。
生产者重要的配置参数解析：
- 设置acks=all。表示必须要有多少个副本确定收到这条消息，之后生产者才会认为这条消息发送成功。这个参数控制着消息持久化的方式有3种类型：
• acks=0。生产者发送消息无需等待kafka服务端的响应。这种方法可靠性最低，即使kafka出现任何异常，producer也无法感知到。
• acks=1。生产者发送消息之后，只需等待其接收的leader副本成功写入就会收到成功的响应，无需等其他follower副本同步写入。这种方式是衡量可靠性和性能之间的一种折中方案。不过依然存在消息丢失的情况，即消息刚写入leader副本并成功返回之后，leader副本崩溃了，那么此时因为其他follower副本都没有同步，那么消息就丢失了。
• acks=all。设置acks=-1是一样的。生产者发送消息之后，需等待ISR（in-sync replica）的所有副本都成功写入才会接收到成功的响应。这种方式是可靠性最强的。不过还需配合另一个参数min.insync.replicas的联动控制。
- 设置retries为一个大于0的值。这个参数用来配置生产者发送消息失败后的重试次数。另外注意，如果设置了retries参数，则建议设置max.in.flight.requests.per.connection=1，不然可能无法保证同一个分区的消息有序性。
max.in.flight.requests.per.connection参数指定了生产者在收到服务端响应之前可以发送多 少条消息，默认为5条。把它设置为1可以保证消息发送的有序性。如果设置大于1，并且配置了重试机制，那么就会出现错序的现象。比如第一条消息写入失败，第二条消息写入成功，那么生产者会重试发送第一条消息，如果此时发送成功，则第一条消息会被插入到第二条消息后面，导致错序。
retry.backoff.ms参数指定两次重试之间的时间间隔，避免无效的频繁重试，默认值为100。
broker服务
当消息来到kafka服务，kafka会通过我们的配置情况，将消息持久化。只有这个操作完成之后，kafka才认为消息「已提交」。那我们看看哪些配置会影响到消息丢失。
-
设置replication.factor>=3。表示主题分区的副本数。生产环境中也是建议至少3个brokers，这样能让每个broker都冗余保存一份数据，保证数据的可靠性。
-
设置min.insync.replicas>1。当acks被设置为all时，这个参数控制指定了消息至少被写入到多少个副本才认为是成功的。min.insync.replocas参数是对acks=all的一种具体约束。但是要确保replication.factor>min.insync.replicas，如果配置两者相等，那么一旦有一个副本挂了，那就是导致消息永远无法发送成功了。
比如配置设置了acks=all, min.insync.replicas=2。
- 一开始副本总数replication.refactor=3，因为acks=all的约束，一条消息需写入3个副本才是成功，因为写入的副本数3大于设置的min.insync.replicas的2，所以没触发下限约束。
- 当一个副本挂了之后，replication.refactor=2，因为acks=all的约束，一条消息需写入2个副本才是成功，因为写入的副本数2等于设置的min.insync.replicas的2，到达下限值但没触发下限约束。
- 当两个副本挂了之后，replication.refactor=1，因为acks=all的约束，一条消息需写入1个副本才是成功，但因为写入的副本数1小于设置的min.insync.replicas的2，所以触发下限制约，写入失败。
-
设置unclean.leader.election.enable=false。这个参数表示是否允许那些没有在ISR（in-sync-replicas）的broker有资格竞选分区leader。默认值为false，建议最好不要主动设置为true。因为如果没有在ISR集合中的副本，可能有些broker副本数据已经落后原先的leader太多了，一旦它成为新的leader副本，那必然出现消息的丢失。
消费端
在讲消费端之后，我们先解释一下上面提到的ISR概念。这里一共是3个概念：
- AR。分区中的所有副本统称为AR（Assigned Replicas）。
- ISR。所有与leader副本保持一定程度同步的副本（包括leader副本在内）称为ISR（In-sync Replicas）。ISR集合是AR的一个子集，并且可以看作是与leader副本数据比较同步的副本。
- OSR。与leader副本同步滞后过多的副本（不包括leader副本）称为OSR（Out-sync Replicas）。OSR集合也是AR的一个子集。OSR是与ISR互斥的。
kafka消费端通过提交自己消费的offset（偏移量）给服务端做记录，这样即使消费者宕机，只要其恢复上线之后便又能从正确的位置拉取新消息进行消费，不会造成消息丢失。下图是官网实例图：
下面我们通过一个例子看下kafka怎样来记录这个offset。下面是一个kafka Log日志文件，这个文件中有9条消息，第一条消息的offset为0（记为LSO，Log Start Offset），最后一条消息的offset为8，最新一条可写入的offset为9（记为LEO，Log End Offset）。
HW（High Watermark）俗称高水位，这里用来指定消费端可消费的最高偏移量，即消费者只能拉取这个HW offset之前的消息进行消费。
LEO（Log End Offset），标识当前日志文件中下一条待写入的消息的offset。
分区ISR集合中的每个副本都会维护自身的LEO，而ISR集合中最小的LEO即为分区的HW，所以对消费者而言只能消费HW之前的消息。上图中，offset在0至5之间的消息，所有指定的副本已同步写入完成，所以消费者可以拉取消费；offset在6～8，说明ISR中部分副本已写入，另一部分处于正在同步状态，所以不能被消费者所拉取。
下面我们拆解一下这个过程。假设某分区的ISR集合中有3个副本，即一个leader副本和2个follower副本，分区的LEO和HW都为3。此时消息3和4从生产者发送之后准备存到leader副本，如下图：
在消息被写入leader副本之后，follower副本便会请求进行同步，如下图：
在同步过程中，follower副本的同步效率和结果都不尽相同，有些已完成全部同步，有些只同步一部分。下图中，所有副本都已同步了消息3，所以HW就往后移动了；而follower2只同步了消息3，未同步消息4，所以HW取当前分区的所有ISR集合中最小的LEO，即HW为4。
等所有副本都成功写入了消息3和4，那么整个分区的HW和LEO都变为5了，因此消费者可以消费到offset小于5之前的所有消息。
从上面这个过程，我们可以清晰得看到kafka消息位移更新的机制。对于消费端来说，维持先消费消息，再提交更新位移，这样可以最大限制保证消息不丢失。
总结
我们从一个消息流开始的生产者发送，到kafka服务存储，再到消费者消费这三大模块分析kafka对于不丢失消息机制的剖析，也包括实践中一些配置参数的解析。当然更多细节和实践，还需我们继续探索。但我相信今天之后，对于kafka的消息不丢失这个问题，我们应该有了些回答的自信！！！
码字不易，一起努力！！！你的点赞是我前行的动力～
近期评论