小惠珠哦

这个屌丝很懒，什么也没留下！

热门标签

各种消息队列经典问题解决方案——消息丢失、顺序消费、消息积压、重复消费_队列挤压又要保证顺序消费如何设计

作者：小惠珠哦 | 2024-06-27 18:00:57

踩

队列挤压又要保证顺序消费如何设计

写在开头：

对于消息队列这种中间件来说，只要进入消息队列就会有几个绕不开的问题，比如：消息丢失 、顺序消费 、消息积压 、重复消费 ，下面就来讲解一下市面上比较常见的各个不同的消息队列产品针对这四个问题的解决方案。

1、Kafka

消息丢失解决方案

对于Kafka这个消息队列来说，消息丢失的环节有下面的几个地方：

1、 消息生产者发送消息给Broker的时候数据丢失

2、Broker异常导致Broker中的数据丢失

3、消息消费者消费异常导致消息丢失

只要我们分别针对这几个不同的环节采取相应的措施就可以有效的防止消息发生丢失，下面就来说一下各个环节的措施：

针对 消息生产者发送消息给Broker的时候数据丢失

首先我们发送消息的流程是将一个消息发送给指定的topic下的分区中，并且在实际的生产环境下我们 一般都是使用多个Broker来保证Kafka的高可用 的，所以一个Topic有多个分区，并且 同一个Topic下的分区会按照主从模式来设置 ，一般是一个Topic下有一个主分区和多个从分区，并且 不同的Topic下的主分区一般要部署在不同的Broker下 ，来进行隔离。

所以我们在消息发送的时候关键的一点就是：保证我们的消息成功发出去了 。所以在进行发送数据的时候Kafka可以来 使用acks参数 来指定写入语义，其中acks有三个值，分别是 acks=0,acks=1,acks=all

acks=0 :最简单的生产者写入语义，表示我 只管发送，发送出去之后至于 broker是否收到、是否持久化、是否主从同步 我不关心。当然这种写入语义是会导致消息丢失的，所以生产中一般不会使用这种写入语义。

acks=1 ：当Broker中的主分区写入成功的时候就认为发送成功了，不关心从分区是否成功同步消息 。这个写入语义也是会导致消息丢失的，比如主分区写入了但是还没有同步给从分区，这时候主分区宕机了，然后从分区中是没有这个消息的。

acks=all : 主分区不仅收到了消息并且完成了分区的主从同步 。这个写入语义相比于上面的两个写入语义更加保险，所以在生产中一般推荐使用这个写入语义。当然这种写入语义的性能可能比上面两种写入语义的 性能差 ，毕竟有得必有失嘛。

补充： ISR：与主分区保持了同步的从分区的数量。比如ISR设置为2，代表着与主分区完成了同步的从分区至少有2个，如果少于两个消息就会发送失败。所以我们在保证生产者发送消息的时候也要配合执行ISR的数量。

但是我们 acks=all的时候一般要配合禁用unClean， unClean是如果ISR中没有任何的分区的时候会选择第一个从分区来作为主分区，那么unClean选举出来的主分区也可能是没有数据的。

顺序消费解决方案

①、Kafka中的 每个分区中的消息是天生保证顺序消费的 ，所以 最简单 的保证顺序消费的方式就是 一个Topic只有一个分区，那么消费者消费这个分区的消息的时候就是顺序消费的。这种 单分区 的方案有两个重要的问题：

性能差，无法应对高并发。因为只有一个分区所以只有一个Broker所以高并发的时候请求都来到了一个Broker上
只有一个分区，所以对应的只有一个消费者，容易造成消息积压

②、Kafka中消息发送的时候消息 生产者可以指定将消息发送给哪个Topic的哪个分区中 。所以我们可以 使用多个分区，只需要保证 同一个业务下的消息发送到同一个分区 就可以了。

但是这个方案还有有两个问题的：

数据可能分配不均匀，产生某些分区有大量的消息，另一些分区消息很少，大量消息的分区可能就会导致 消息积压
针对 数据分布不均匀 的问题，常见的方案就是使用 一致性哈希算法 。我们将分区做成一个哈希环，然后我们根据消息的业务标识去计算哈希值来决定分配到哪个分区，同时我们是可以根据分区数据的分配情况来调整分区的数量来让数据分配均匀的。
增加分区后可能会导致 **消息乱序 **

一个最基本的方案就是让 新增的分区先暂停消费一段时间 ，让原来的老分区先消费完毕之后再消费。

上面说的两种方法针对的都是同一个Topic下的消息有序，如果设计到多个不同的Topic下分区消息有序的话，那么需要引入一个协调者。

③、本地内存队列+多线程消费 ：

可以使用一个消费者来将Kafka对应分区中的消息取出来，然后根据业务标识 分配到不同的内存队列中 ，然后不同的线程去自己对应的队列中执行任务。这样也可以保证同一业务下的消息有序性。

消息积压解决方案

①、增加消费者的数量和分区数量一样多

②、增加分区数量

③、如果公司不允许增加分区的话，可以创建一个新的Topic，这个Topic下有更多的分区，然后让新的消息发送到这个新的Topic中，同时创建一个消费者去消费老Topic分区中的数据，只是负责将老Topic分区中的数据转发到新的Topic的分区中。

④、优化消费者的性能

使用更好的机器
对消费者的业务逻辑进行优化

⑤、批量消费+多线程 ：消费者一次从队列中拉去一批比如10个消息进行处理，然后交给10个线程并发执行然后批量提交。

但是这个方案有两个问题：1. 重复消费 2. 部分失败

针对重复消费的问题我们可以使用幂等性来处理。

针对部分失败的问题： 1. 失败后立即重试，但是要注意设置重试的次数和重试时间 2. 使用一个异步线程来进行重试 3. 将失败的消息重新丢回到消息队列中，但是要注意标记失败次数，当达到一个最大失败次数之后就不重试了，需要人工干预了。

重复消费解决方案

解决重复消费的方案就是保证接口的 幂等性 ，即使多次消费同一个消息，那么对应的后续的逻辑也只会执行一次。

①、使用唯一索引

使用唯一索引应该是最简单的保证幂等性的方法。但是使用唯一索引来实现幂等性的时候要注意两个地方：

要将插入唯一索引和业务处理放在 同一个本地事务 中。不然可能会出现成功插入了唯一索引但是业务处理失败了，那么之后这个消息就再也得不到执行了。
如果因为种种原因不能使用本地事务，那么就需要考虑 最终一致性 方案了，这个时候就需要引入一个第三方了，比如我们可以引入一个 定时任务 。流程如下：我们先插入数据到唯一索引中，同时唯一索引对应的数据设置为 初始状态 ；然后进行业务处理；如果业务成功处理那么就将唯一索引对应的数据设置为 成功状态 ；同时我们的异步任务会 定时扫描 唯一索引中为初始状态的数据，然后和业务处理表中的数据进行对比，如果发现业务表中成功处理了但是唯一索引表中的状态还是初始状态，那么就将唯一索引中的数据状态变成成功。反正可以触发 重试机制 。

但是使用唯一索引的方案有一个缺点就是： 性能瓶颈在数据库上，无法支持高并发

②、布隆过滤器+Redis+唯一索引

大概的流程如下：1. 请求来了之后先经过布隆过滤器判断，如果布隆过滤器判断这个唯一标识不存在说明这个消息没有被处理过，那么就可以直接执行；如果这个布隆过滤器判断存在那么有一定的概率出现误判，所以我们又加了一个Redis;

这个Redis中存放了最近处理的消息的唯一标识，然后判断如果Redis中存在这个消息的唯一标识，那么说明这个消息已经处理过了，那么就不会进行处理，如果Redis中不存在则会进入到数据库层；

进入到数据库层面后利用唯一索引来判断这个消息是不是处理过了，如果插入失败说明已经消费过了，则不进行处理；如果插入成功说明没有消费过则执行消息的消费逻辑。

2、RabbitMQ

消息丢失解决方案

针对消息生产者是否成功叫消息发送给交换机：

可以使用RabbitMQ中的 confirm机制 ，如果RabbitMQ成功接收到消息之后会给消息生产者返回一个confirm，当消息生产者接收到confirm的时候就可以判定RabbitMQ已经成功接收到消息了，消息成功发送了，反之说明消息可能没有成功发送，需要进行重试发送。

还可以使用RabbitMQ提供的 事务机制 ，在发送消息的时候做成一个事务，来保证发送的消息要么全部成功，要么全部失败。

注意：confirm机制和事务机制，只能同时使用一个，不能一起使用。

针对RabbitMQ服务器宕机导致RabbitMQ中的消息丢失：

开启RabbitMQ的 持久化 功能，在交换机中进行持久化，同时也在队列中进行持久化。

针对消费者消费消息的时候发生了异常，导致消息从RabbitMQ中丢失：

关闭RabbitMQ的自动ack机制，使用 手动ACK机制 ，当消费者确实成功消费后给RabbitMQ一个ACK信号，然后RabbitMQ可以根据配置来决定是不是要将这个消息删除。

顺序消费解决方案

①、使用多个队列 ：每个业务对应一个队列，将对应的业务中的消息发送给对应的队列，然后消费者消费自己对应的业务队列即可

②、本地内存队列+多线程消费 ：使用一个队列，然后消费者从队列中将消息取出来，然后根据业务标识将消息放入到不同的内存队列中，然后不同的线程消费自己对应的内存队列。

消息积压解决方案

①、对于RabbitMQ来说，消息积压问题比较好解决，因为一个队列可以对应有多个消费者，多个消费者轮询消费队列中的消息，所以一旦出现了消息积压，我们最简单的方式就是可以 增加消费者的数量 。

②、使用优先级队列 ：我们有些任务可能有比较高的优先级，我们对实时性要求比较高，那么我们可以使用优先级队列将优先级高的任务发到优先级队列中，让优先级高的消息优先得到消费

③、优化消费者的性能

使用更好的机器
对消费者的业务逻辑进行优化

重复消费解决方案

对于三种消息队列的重复消费解决方案其实都一样，都是保证接口的 幂等性 即可。

解决重复消费的方案就是保证接口的幂等性，即使多次消费同一个消息，那么对应的后续的逻辑也只会执行一次。

①、使用唯一索引

使用唯一索引应该是最简单的保证幂等性的方法。但是使用唯一索引来实现幂等性的时候要注意两个地方：

要将插入唯一索引和业务处理放在同一个本地事务中。不然可能会出现成功插入了唯一索引但是业务处理失败了，那么之后这个消息就再也得不到执行了。
如果因为种种原因不能使用本地事务，那么就需要考虑最终一致性方案了，这个时候就需要引入一个第三方了，比如我们可以引入一个定时任务。流程如下：我们先插入数据到唯一索引中，同时唯一索引对应的数据设置为初始状态；然后进行业务处理；如果业务成功处理那么就将唯一索引对应的数据设置为成功状态；同时我们的异步任务会定时扫描唯一索引中为初始状态的数据，然后和业务处理表中的数据进行对比，如果发现业务表中成功处理了但是唯一索引表中的状态还是初始状态，那么就将唯一索引中的数据状态变成成功。反正可以出发重试机制。

但是使用唯一索引的方案有一个缺点就是：性能瓶颈在数据库上，无法支持高并发

②、布隆过滤器+Redis+唯一索引

3、RocketMQ

消息丢失解决方案

针对消息生产者将消息投递给RocketMQ的时候发生消息丢失：

消息生产者可以选择使用 同步发送 ，消息发出去后会 阻塞等待 Broker确认成功接收到消息。

针对RocketMQ宕机发生的消息丢失的情况：

开启RocketMQ的 持久化 功能，将RocketMQ中的消息进行持久化；RocketMQ使用 主从架构 来进行部署。

针对RocketMQ消费者消费消息失败后导致的消息丢失的情况：

开启 手动ACK机制 ，当消费者成功将消息消费后给RocketMQ发送一个ACK消息，只有当RocketMQ接收到后才会确认消息是被成功消费了。

顺序消费解决方案

RocketMQ中生产者在给Topic发送消息的时候，默认使用轮询机制给该Topic下的所有队列进行发送消息。如果我们要保证有序的话，我们可以使用 哈希+取模 的算法来针对同一个业务的业务标识来计算 将同一个业务的消息发送到同一个队列中 ，同一个队列中的消息是天生有序消费的。

消息积压解决方案

①、增加消费者的数量 ：因为一个队列只能有一个消费者，所以一般也需要增加更多的队列数量

②、使用批量处理，RocketMQ是支持批量消息的

③、优化消费者的性能

使用更好的机器
对消费者的业务逻辑进行优化

重复消费解决方案

对于三种消息队列的重复消费解决方案其实都一样，都是保证接口的幂等性即可。

解决重复消费的方案就是保证接口的幂等性，即使多次消费同一个消息，那么对应的后续的逻辑也只会执行一次。

①、使用唯一索引

使用唯一索引应该是最简单的保证幂等性的方法。但是使用唯一索引来实现幂等性的时候要注意两个地方：

要将插入唯一索引和业务处理放在同一个本地事务中。不然可能会出现成功插入了唯一索引但是业务处理失败了，那么之后这个消息就再也得不到执行了。
如果因为种种原因不能使用本地事务，那么就需要考虑最终一致性方案了，这个时候就需要引入一个第三方了，比如我们可以引入一个定时任务。流程如下：我们先插入数据到唯一索引中，同时唯一索引对应的数据设置为初始状态；然后进行业务处理；如果业务成功处理那么就将唯一索引对应的数据设置为成功状态；同时我们的异步任务会定时扫描唯一索引中为初始状态的数据，然后和业务处理表中的数据进行对比，如果发现业务表中成功处理了但是唯一索引表中的状态还是初始状态，那么就将唯一索引中的数据状态变成成功。反正可以出发重试机制。

但是使用唯一索引的方案有一个缺点就是：性能瓶颈在数据库上，无法支持高并发

②、布隆过滤器+Redis+唯一索引

END…

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小惠珠哦/article/detail/763291