18消息投递如何保证消息仅仅被消费一次？

　　上节课我们通过在电商系统增加消息组件达到了削峰填谷的作用，对次要业务逻辑做了异步处理，对不同的系统模块做了解耦。因为业务逻辑从队列中移除了，所以我们要有响应的队列处理逻辑了，这时系统架构就变成了如下：
　　今天再来讨论一个非主流程的业务逻辑，比如客户购买了一个商品之后，会给用户发红包，但是发红包这个流程是非必要的业务关键流程，所以可以通过消息队列来异步处理。但是这里面容易出现一个问题，如果红包消息重复处理了就会导致公司利益损失，这个问题怎么解决呢？这个就是我们今天的话题：消息的重复消费问题。
　　那么我们如何来保证产生的消息一定会被消费并且只被消费一次呢？消息为什么会丢失
　　如果消息只被消费一次，首先保证消息不丢失。那么消息从被写入消息队列到被消费者完成，这个链路会有哪些丢失消息的可能，三个方面消息从生产者写入消息队列的过程消息在消息队列中的存储场景消息被消费者消费的过程
　　1、在消息生产的过程中丢失消息
　　首先消息生产者是我们的业务服务器，消息队列服务器为单独部署的。虽然是内网，但这个过程也有可能出现网络抖动导致消息丢失的可能。
　　针对这种情况，建议采用消息重发机制，但是也不能无限制的重发，如果不是因为消息队列发生故障或是网络断开，一般重试23次即可解决。而重发机制，也是大多数消息队列本身提供的功能。
　　消息重发存在的问题就是消息的重复消费，现实的意义就是你收到了两个红包，公司就亏损了。
　　那么消息发送到了消息队列是不是就万无一失了呢？也不是，消息队列中仍然有丢失的可能。2、在消息队列中丢失消息
　　已kafka为例，消息在kafka中是存在本地磁盘上的，而为了减少消息存储对磁盘的随机IO，一般消息会先写入到操作系统的pagecache中，然后再找合适的时间刷新到磁盘上。
　　比如，卡夫卡可以设置一定的时间或者消息累积到一定量时候再刷盘，也就是所说的异步刷盘。
　　不过如果发生机器断电或者机器重启，pagecache中还没来得及刷盘消息就会丢失，那怎么解决呢？
　　你可以设置把刷盘时间设置很短或者消息累计很少就刷，但是这样频繁刷盘对性能也有很大影响，经验来看，出现机器宕机或者掉电的几率也不高，所以不建议这么做。
　　如果系统对消息丢失容忍度很低，可以考虑集群部署消息服务，通过部署多个副本保证消息尽量不丢失。
　　kafka集群中还有一个Leader负责消息的写入和消费，还有多个follower负责消息的数据备份，Follower中还有一个特殊的集合ISR（insyncreplicas）当Leader故障时，新的Leader会从ISR中选择，默认Leader的数据会异步的复制给Follower，这样在Leader发生掉电或者宕机时，Kafka会从Follower中消费消息，减少消息丢失的可能。
　　由于消息默认是从Leader复制到Follower的，所以一旦Leader宕机，那么没来得及复制的消息还是会丢失，为了解决这个问题，Kafka提供了一种机制：acks，当这个选项被设置为all时，生产者发送的每一条消息除了发给Leader外还会发给所有的ISR，并且必须得到Leader和所有ISR的确认后才被认为发送成功。这样，只有Leader和所有的ISR都挂了消息才会丢失。
　　从上面这张图来看，当设置acksall时，需要同步执行1、3、4三个步骤，对于消息生产的性能来说也是有比较大的影响的，所以你在实际应用中需要仔细地权衡考量。我给你的建议是：如果你的消息一条不能丢失，建议不要开启同步刷盘，而是用集群方式解决，可以配置当所有ISRFollower都接收到消息才返回成功如果对消息丢失有一定容忍度，不建议使用集群，或者即使是集群也配置一个Follower同步成功即返回成功。上面我们发红包的业务就有一定容忍度，如果消息丢失可以通过补发来实现。3、在消费的过程中存在消息丢失的可能
　　一个消息消费的进度是记录在消息队列群里的，消费的过程分为三步：接收消息、处理消息、更新消费进度。
　　接受消息和处理消息都有可能发生失败，比如接收消息发生网络抖动，导致消息没有正确接收到，处理消息可能发生一些异常，这是入股更新了消费进度，那么这条消息就永远不会被处理了。
　　所以你这里需要注意了，一定要等到消息正确处理完成再正确提交，这时也有一个问题就是，加入消息已经正确处理，但是由于网络问题没有正确提交消费进度，这个时候就会导致消息会重复的消费。如何保证消息只被消费一次
　　从上面分析可以得出，为了避免消息丢失，我们做了两方面的努力：一是需要牺牲一部分性能，而是可能造成消息的重复消费。
　　性能损耗还可以，一般业务系统只有写操作才会发送消息队列，一般系统的写请求并发量不高，但是消息一旦被重复消费就会造成业务逻辑的错误。那么如何避免呢？
　　完全的做到不消费重复消息很难做到，所以我们转换思路，只要保证即使消费到了重复消息，从消费结果来看，跟只消费一次的等同就好了。也就是保证生产和消费的过程幂等！！！
　　说白了，就是一件事无论做了多少次，产生的结果都是一样的，这个就是幂等了。在生产、消费过程中增加消息幂等性的保证
　　消息的生产和消费过程都有可能产生重复，所以要做的就是生产和消费的过程中增加消息的幂等性，这样就可以认为从结果上看实际只消费了一次。
　　在消息生产过程中，在Kafka0。11版本和Pulsar中都支持produceridempotency的特性，翻译过来就是生产过程的幂等性，这种特性保证消息虽然可能在生产端产生重复，但是最终在消息队列存储时只会存储一份
　　它的做法是给每一个生产者一个唯一的ID，并且为生产的每一条消息赋予一个唯一ID，消息队列的服务端会存储生产者ID，最后一条消息ID的映射。当某一个生产者产生新的消息时，消息队列服务端会比对消息ID是否与存储的最后一条ID一致，如果一致就认为是重复的消息，服务端会自动丢弃。
　　而在消费端，幂等性的保证会稍微复杂一些，你可以从通用层和业务层两个层面来考虑。
　　通用层面，你可以在消息被生产的时候使用发号器给生成一个唯一id，消息被处理后，把这个id存储在数据库中，在处理下一条消息之前从数据库查询这个id消息是否被消费过，如果被消费就放弃消费。
　　你可以看到，无论是生产端的幂等性保证方式还是消费端通用的幂等性保证方式，它们的共同特点都是为每一个消息生成一个唯一的ID，然后在使用这个消息的时候，先比对这个ID是否已经存在，如果存在则认为消息已经被使用过。所以这种方式是一种标准的实现幂等的方式，你在项目之中可以拿来直接使用，它在逻辑上的伪代码就像下面这样：booleanisIDExistedselectByID（ID）；判断ID是否存在if（isIDExisted）｛return；存在则直接返回｝else｛process（message）；不存在，则处理消息saveID（ID）；存储ID｝
　　这样存在一个问题，如果消息处理完成之后，还没来得及写入数据库，消费者宕机了，重启之后发现数据库没有，还是会重复消费消息，这个时候就考虑引入事务机制，保证消息的消费跟数据库的写入必须同时成功或者失败，但是这样的成本会很高，所以如果要求不是很严格，可以直接使用这种通用的方案，不要考虑引入事务。
　　在业务层面怎么处理呢？这里有很多种处理方式，其中有一种是增加乐观锁的方式。比如你的消息处理程序需要给一个人的账号加钱，那么你可以通过乐观锁的方式来解决。
　　具体的操作方式是这样的：你给每个人的账号数据中增加一个版本号的字段，在生产消息时先查询这个账户的版本号，并且将版本号连同消息一起发送给消息队列。消费端在拿到消息和版本号后，在执行更新账户金额SQL的时候带上版本号，类似于执行：
　　你看，我们在更新数据时给数据加了乐观锁，这样在消费第一条消息时，version值为1，SQL可以执行成功，并且同时把version值改为了2；在执行第二条相同的消息时，由于version值不再是1，所以这条SQL不能执行成功，也就保证了消息的幂等性。