kafka在高并发的情况下,如何避免消息丢失和消息重复?

思考：

1，消息丢失是什么造成的，从生产端和消费端两个角度来考虑？

2，消息重复是什么造成的，从生产端和消费端两个角度来考虑？

3，如何保证消息有序？

4，如果保证消息不重不漏，损失的是什么？

1.为什么会发生消息丢失和消息重复？
消息发送
Kafka消息发送有两种方式：同步（sync）和异步（async），默认是同步方式，可通过producer.type属性进行配置。Kafka通过配置request.required.acks属性来确认消息的生产：
0—表示不进行消息接收是否成功的确认；
1—表示当Leader接收成功时确认；
-1—表示Leader和Follower都接收成功时确认；

综上所述，有6种消息生产的情况，下面分情况来分析消息丢失的场景：
（1）acks=0，不和Kafka集群进行消息接收确认，则当网络异常、缓冲区满了等情况时，消息可能丢失；
（2）acks=1、同步模式下，只有Leader确认接收成功后但挂掉了，副本没有同步，数据可能丢失；

消息消费
Kafka消息消费有两个consumer接口，Low-level API和High-level API：
Low-level API：消费者自己维护offset等值，可以实现对Kafka的完全控制；
High-level API：封装了对parition和offset的管理，使用简单；

如果使用高级接口High-level API，可能存在一个问题就是当消息消费者从集群中把消息取出来、并提交了新的消息offset值后，还没来得及消费就挂掉了，那么下次再消费时之前没消费成功的消息就“诡异”的消失了；
解决办法：
针对消息丢失：同步模式下，确认机制设置为-1，即让消息写入Leader和Follower之后再确认消息发送成功；异步模式下，为防止缓冲区满，可以在配置文件设置不限制阻塞超时时间，当缓冲区满时让生产者一直处于阻塞状态；
针对消息重复：将消息的唯一标识保存到外部介质中，每次消费时判断是否处理过即可。

消息丢失解决方案:
首先对kafka进行限速， 其次启用重试机制，重试间隔时间设置长一些，最后Kafka设置acks=all，即需要相应的所有处于ISR的分区都确认收到该消息后，才算发送成功

消息重复解决方案
消息可以使用唯一id标识
生产者（ack=all 代表至少成功发送一次)
消费者 （offset手动提交，业务逻辑成功处理后，提交offset）
落表（主键或者唯一索引的方式，避免重复数据）
业务逻辑处理（选择唯一主键存储到Redis或者mongdb中，先查询是否存在，若存在则不处理；若不存在，先插入Redis或Mongdb,再进行业务逻辑处理）

使用同步模式的时候，有3种状态保证消息被安全生产，在配置为1（只保证写入leader成功）的话，如果刚好leader partition挂了，数据就会丢失。还有一种情况可能会丢失消息，就是使用异步模式的时候，当缓冲区满了，如果配置为0（还没有收到确认的情况下，缓冲池一满，就清空缓冲池里的消息），数据就会被立即丢弃掉。

在数据生产时避免数据丢失的方法：
只要能避免上述两种情况，那么就可以保证消息不会被丢失。就是说在同步模式的时候，确认机制设置为-1，也就是让消息写入leader和所有的副本。还有，在异步模式下，如果消息发出去了，但还没有收到确认的时候，缓冲池满了，在配置文件中设置成不限制阻塞超时的时间，也就说让生产端一直阻塞，这样也能保证数据不会丢失。

在数据消费时，避免数据丢失的方法：如果使用了storm，要开启storm的ackfail机制；如果没有使用storm，确认数据被完成处理之后，再更新offset值。低级API中需要手动控制offset值。

数据重复消费的情况，如果处理？
（1）去重：将消息的唯一标识保存到外部介质中，每次消费处理时判断是否处理过；
（2）不管：大数据场景中，报表系统或者日志信息丢失几条都无所谓，不会影响最终的统计分析结果。

下面是文章详情，这里先简单总结一下：

消费端重复消费：很容易解决，建立去重表

消费端丢失数据：也容易解决，关闭自动提交offset，处理完之后受到移位

生产端重复发送：这个不重要，消费端消费之前从去重表中判重就可以

生产端丢失数据：这个是最麻烦的情况

解决策略：

1.异步方式缓冲区满了，就阻塞在那，等着缓冲区可用，不能清空缓冲区

2.发送消息之后回调函数，发送成功就发送下一条，发送失败就记在日志中，等着定时脚本来扫描

（发送失败可能并不真的发送失败，只是没收到反馈，定时脚本可能会重发）

如何保证有序：

如果有一个发送失败了，后面的就不能继续发了，不然重发的那个肯定乱序了

生产者在收到发送成功的反馈之前，不能发下一条数据，但我感觉生产者是一个流，阻塞生产者感觉业务上不可行，怎么会因为一条消息发出去没收到反馈，就阻塞生产者

同步发送模式：发出消息后，必须阻塞等待收到通知后，才发送下一条消息

异步发送模式：一直往缓冲区写，然后一把写到队列中去

两种都是各有利弊：

同步发送模式虽然吞吐量小，但是发一条收到确认后再发下一条，既能保证不丢失消息，又能保证顺序

============================下面是原文====================================

Kafka消息保证生产的信息不丢失和重复消费问题
1）使用同步模式的时候，有3种状态保证消息被安全生产，在配置为1（只保证写入leader成功）的话，如果刚好leader partition挂了，数据就会丢失。
2）还有一种情况可能会丢失消息，就是使用异步模式的时候，当缓冲区满了，如果配置为0（还没有收到确认的情况下，缓冲池一满，就清空缓冲池里的消息），
数据就会被立即丢弃掉。

在数据生产时避免数据丢失的方法：
只要能避免上述两种情况，那么就可以保证消息不会被丢失。
1）就是说在同步模式的时候，确认机制设置为-1，也就是让消息写入leader和所有的副本。
2）还有，在异步模式下，如果消息发出去了，但还没有收到确认的时候，缓冲池满了，在配置文件中设置成不限制阻塞超时的时间，也就说让生产端一直阻塞，这样也能保证数据不会丢失。
在数据消费时，避免数据丢失的方法：如果使用了storm，要开启storm的ackfail机制；如果没有使用storm，确认数据被完成处理之后，再更新offset值。低级API中需要手动控制offset值。

消息队列的问题都要从源头找问题，就是生产者是否有问题。

讨论一种情况，如果数据发送成功，但是接受response的时候丢失了，机器重启之后就会重发。

重发很好解决，消费端增加去重表就能解决，但是如果生产者丢失了数据，问题就很麻烦了。

数据重复消费的情况，如果处理
（1）去重：将消息的唯一标识保存到外部介质中，每次消费处理时判断是否处理过；
（2）不管：大数据场景中，报表系统或者日志信息丢失几条都无所谓，不会影响最终的统计分析结

Kafka到底会不会丢数据(data loss)? 通常不会，但有些情况下的确有可能会发生。下面的参数配置及Best practice列表可以较好地保证数据的持久性(当然是trade-off，牺牲了吞吐量)。笔者会在该列表之后对列表中的每一项进行讨论，有兴趣的同学可以看下后面的分析。
没有银弹，如果想要高吞吐量就要能容忍偶尔的失败（重发漏发无顺序保证）。

block.on.buffer.full = true
acks = all
retries = MAX_VALUE
max.in.flight.requests.per.connection = 1
使用KafkaProducer.send(record, callback)
callback逻辑中显式关闭producer：close(0) 
unclean.leader.election.enable=false
replication.factor = 3 
min.insync.replicas = 2
replication.factor > min.insync.replicas
enable.auto.commit=false
消息处理完成之后再提交位移
给出列表之后，我们从两个方面来探讨一下数据为什么会丢失：

1. Producer端

目前比较新版本的Kafka正式替换了Scala版本的old producer，使用了由Java重写的producer。新版本的producer采用异步发送机制。KafkaProducer.send(ProducerRecord)方法仅仅是把这条消息放入一个缓存中(即RecordAccumulator，本质上使用了队列来缓存记录)，同时后台的IO线程会不断扫描该缓存区，将满足条件的消息封装到某个batch中然后发送出去。显然，这个过程中就有一个数据丢失的窗口：若IO线程发送之前client端挂掉了，累积在accumulator中的数据的确有可能会丢失。

Producer的另一个问题是消息的乱序问题。假设客户端代码依次执行下面的语句将两条消息发到相同的分区
producer.send(record1);
producer.send(record2);
如果此时由于某些原因(比如瞬时的网络抖动)导致record1没有成功发送，同时Kafka又配置了重试机制和max.in.flight.requests.per.connection大于1(默认值是5，本来就是大于1的)，那么重试record1成功后，record1在分区中就在record2之后，从而造成消息的乱序。很多某些要求强顺序保证的场景是不允许出现这种情况的。

发送之后重发就会丢失顺序

鉴于producer的这两个问题，我们应该如何规避呢？？对于消息丢失的问题，很容易想到的一个方案就是：既然异步发送有可能丢失数据， 我改成同步发送总可以吧？比如这样：

producer.send(record).get();
这样当然是可以的，但是性能会很差，不建议这样使用。因此特意总结了一份配置列表。个人认为该配置清单应该能够比较好地规避producer端数据丢失情况的发生：(特此说明一下，软件配置的很多决策都是trade-off，下面的配置也不例外：应用了这些配置，你可能会发现你的producer/consumer 吞吐量会下降，这是正常的，因为你换取了更高的数据安全性)

block.on.buffer.full = true  尽管该参数在0.9.0.0已经被标记为“deprecated”，但鉴于它的含义非常直观，所以这里还是显式设置它为true，使得producer将一直等待缓冲区直至其变为可用。否则如果producer生产速度过快耗尽了缓冲区，producer将抛出异常。缓冲区满了就阻塞在那，不要抛异常，也不要丢失数据
acks=all  很好理解，所有follower都响应了才认为消息提交成功，即"committed"
retries = MAX 无限重试，直到你意识到出现了问题
max.in.flight.requests.per.connection = 1 限制客户端在单个连接上能够发送的未响应请求的个数。设置此值是1表示kafka broker在响应请求之前client不能再向同一个broker发送请求。注意：设置此参数是为了避免消息乱序
使用KafkaProducer.send(record, callback)而不是send(record)方法   自定义回调逻辑处理消息发送失败，比如记录在日志中，用定时脚本扫描重处理
callback逻辑中最好显式关闭producer：close(0) 注意：设置此参数是为了避免消息乱序（仅仅因为一条消息发送没收到反馈就关闭生产者，感觉代价很大）
unclean.leader.election.enable=false   关闭unclean leader选举，即不允许非ISR中的副本被选举为leader，以避免数据丢失
replication.factor >= 3   这个完全是个人建议了，参考了Hadoop及业界通用的三备份原则
min.insync.replicas > 1 消息至少要被写入到这么多副本才算成功，也是提升数据持久性的一个参数。与acks配合使用
保证replication.factor > min.insync.replicas  如果两者相等，当一个副本挂掉了分区也就没法正常工作了。通常设置replication.factor = min.insync.replicas + 1即可

2. Consumer端
　　consumer端丢失消息的情形比较简单：如果在消息处理完成前就提交了offset，那么就有可能造成数据的丢失。由于Kafka consumer默认是自动提交位移的，所以在后台提交位移前一定要保证消息被正常处理了，因此不建议采用很重的处理逻辑，如果处理耗时很长，则建议把逻辑放到另一个线程中去做。为了避免数据丢失，现给出两点建议：
enable.auto.commit=false  关闭自动提交位移
在消息被完整处理之后再手动提交位移
————————————————
版权声明：本文为CSDN博主「千淘万漉」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/matrix_google/article/details/79888144