使用消息中间件时，如何保证消息仅仅被消费一次？

消息中间件使用广泛，常用来削峰填谷、系统解耦、异步处理。异步处理可能是使用的最多的场景了，比如现在的技术博客网站，都采用积分制，用户发表一篇文章后，可以获取想要的积分，为了提升系统的性能，给用户加积分的操作可以异步处理，并不需要放在同步流程中。

我们可以把用户ID，需要增加的积分封装成一条消息投递到消息系统中，异步处理加积分操作，由于这是发生在不同服务器之间，消息有可能投递失败、处理失败等问题，从而导致用户加积分失败，还有一种可能是消息重复投递，那么用户就有可能重复加积分，不管出现那种情况，都是不正常的情况。

要避免上面的两种情况，就需要我们尽量保证消息不丢失和消息只被消费一次，这篇文章抛开具体的消息中间件，从消息系统的通用层面上，谈谈如何避免这两种情况。

1、保证消息不丢失

一条消息从生产到消费这条链路中，有三个地方可能会造成消息丢失，分别如下：

• 消息从生产者写入到消息队列的过程投递失败。

• 消息在消息队列中，持久化失败。

• 消息被消费者消费的过程出现异常。

在消息生产的过程中投递失败

消息生产者和消息系统一般都是独立部署在不同的服务器上，两台服务器之间要通信就要通过网络来完成，网络是不稳定，可能会发生抖动，那么数据就有可能丢失。网络发生抖动会有以下两种情况。

在消息生产的过程中丢失消息

情景一：消息在传送给消息系统的过程中发生网络抖动，数据直接丢失。

情景二：消息已经到达消息系统，但是在消息系统给生产者服务器返回信息时，网络发生抖动，此时的数据不一定真正的丢失，很可能只是生产者认为数据丢失。

针对消息在消息生产时丢失，可以采取重投机制，当程序检测到网络异常时，将消息再次投递到消息系统。但是重新投递在情景二情况下，可能造成数据重复，如何解决这个问题，在后面会提到。

在消息队列中持久化失败

消息系统是可以对消息进行持久化，一般都是将消息存储到本地磁盘中，当然也有少数消息中间件支持将数据持久化到数据库中，那么消息系统的性能可能就会下降。

如果你对 Redis 的持久化有一定的了解话，你会发现 Redis 在持久化数据时并不是每新增一条就立即存入到本地磁盘，而是会将数据先写入到操作系统的 Page Cache 中，当满足一定条件时，再将 Page Cache 中的数据刷入磁盘，因为这样可以减少对磁盘的随机 I/O 操作，我们知道随机 I/O 是非常耗时的，这样也提高了系统性能，消息中间件也不例外，在持久化时也是采用这种方式。

在某些极端情况下，可能会造成 Page Cache 中的数据丢失，比如突然停电或者机器异常重启操作。要解决 Page Cache 中数据丢失问题，可以采用集群部署的方式，来尽量保证数据不丢失。

在消费的过程中存在消息丢失

消息在消费过程中也是会发生丢失的，而且在消费过程中丢失的概率要比前两种情况大很多。一条消息消费过程大概分成三步：消费者拉取消息，消费者处理消息，消息系统更新消费进度。

图片描述

第一步在拉取消息的时候可能发生网络抖动异常，第二步在处理消息的时候可能发生一些业务异常，而导致流程并没有走完，如果在第一步、第二步发生异常的情况下，通知消息系统更新消费进度，那么这条失败的消息就永远不会在被处理了，自然就丢失了，其实我们的业务并没有跑完。

要避免消息在消费时丢失的情况，可以在消息接收和处理完成之后才更新消费进度，但是在极端的情况下，会出现消息重复消费的问题，比如某一条消息在处理完成之后，消费者宕机了，这时还没有更新消费进度，消费者重启后，这条消息还是会被消费到。

2、保证消息只被消费一次

消息系统本身不能保证消息仅被消费一次，因为消费本身可能重复、下游系统启动拉取重复、失败重试带来的重复、补偿逻辑导致的重复都有可能造重复消息， 要保证消息仅被消费一次可以利用等幂性来实现 。

等幂是数学上的一个概念，就是多次执行同一个操作和执行一次操作，最终得到的结果是相同的。

从等幂的概念上就可以看出来，就算消息执行多次也不会对系统造成影响，那么在使用消息系统时如何保证等幂性呢？因为生产者和消费者都有可能产生重复消息，所以要在生产者和消费者两端都保证等幂性。

保证生产者等幂性，在生产消息的时候，利用雪花算法给消息生成一个全局 ID，在消息系统中维护消息已 ID 映射关系，如果在映射表中已经存在相同 ID，这丢弃这条消息，虽然消息被投递了两次，但是实际上就保存了一条，避免了消息重复问题。

生产者等幂性跟所选者的消息中间件有关系，因为绝大数情况下消息系统不需要我们自己实现，所以等幂性是不太好控制的， 消费者等幂性才是我们开发人员控制的重点方向 。

在消费者端可以从通用层和业务层两个方面来做等幂操作，取决于我们的业务要求。

在通用层面中，利用好消息生成是产生的全局唯一ID，消息被处理成功后，把这个全局 ID 存入到数据中，在处理下一条消息之前，先从数据库中查询这个全局 ID 是否存在，如果已经存在，则直接放弃该消息。

利用这个全局唯一ID就实现了消息等幂性，伪代码如下：

boolean isIDExisted = selectByID(ID); // 判断ID是否存在
if(isIDExisted) {
  return; //存在则直接返回
} else {
  process(message); //不存在，则处理消息
  saveID(ID);   //存储ID
}

但是在极端情况下，这种方式还是会出问题，如果消息在处理之后，还没来得及保存到数据库，消费者就宕机重启了，重启之后还会再次获取该消息，执行时查询该消息并未被消费过，还是会执行两次消费。可以引入数据库事务来解决这个问题，但是会降低系统性能。如果对消息重复消费没有特别严格要求的话，直接使用这种没有引入事务的通用方案就好了，毕竟这也是极小概率的事情。

在业务层面上，我们可选择性就变多了，比如乐观锁、悲观锁、内存去重（https://github.com/RoaringBitmap/RoaringBitmap）等方法。

我们拿乐观锁来举例，比如我们要给一个用户加积分，因为加积分操作并不需要放在主业务中，所以就可以使用消息系统来异步通知，要使用乐观锁，就需要给积分表添加一个版本号字段。并且在生产消息的时候先查询这个账号的版本号并且连同消息一起发送到消息系统中。

图片描述

消费者拿到消息和版本号后，在执行更新积分操作的 SQL 时带上版本号，类似于：

update score set score = score + 20, version=version+1 where userId=1 and version=1;

这条消息消费成功后，version 就变成了 2，那么如果有重复的 version=1 的消息再次被消费者拉取到，SQL 语句并不会执行成功，从而保证了消息的幂等性。

要保证消息仅被消费一次，我们需要把重点放在消费者这一段，利用等幂性来保证消息被消费一次。

今天站在消息中间件的通用层面上，聊了聊如何保证数据不丢失和仅被消费一次，希望今天的文章对您的学习或者工作有所帮助，如果您认为文章有价值，欢迎点个赞，谢谢。