Kafka - 延迟消息的实现方案(附:RabbitMQ延迟消息)
作者:hangge | 2024-06-04 08:18
一、基本介绍
1,延迟队列
延迟队列是一种特殊的队列。它里面的每个元素都有一个过期时间:
- 当元素还没到过期时间的时候,如果我们试图从队列里面获取一个元素,我们会被阻塞。
- 当有元素过期的时候,我们就会拿到这个过期的元素。也可以这样想,我们拿到的永远是最先过期的那个元素。
提示:很多语言本身就提供了延迟队列的实现,比如说在 Java 里面的 DelayQueue。
2,延迟消息
基于消息队列的延迟队列,也叫做延迟消息。具体来说,延迟消息是指消息不是立刻被消费的,而是在经过一段时间之后,才会被消费。在到时间之前,这个消息一直都被存储在消息队列的服务器上。注意:RabbitMQ 有插件支持延迟消息功能,而 RocketMQ 和 Kafka 则只能自己研发。
二、RabbitMQ 实现延迟消息
1,使用插件实现
(1)RabbitMQ 有一个延迟消息的插件 rabbitmq_delayed_message_exchange,只需要启用这个插件就可以使用延迟消息。这个插件的基本原理也比较简单,就是实现了一个 exchange。这个 exchange 控制住了消息什么时候会被真的投递到队列里。- 如下图所示,消息会先暂时存储在 exchange 里面。它使用的是 Mnesia 来存储。如果你不知道 Mnesia 是什么,就直观地把它看作一个基于文件的数据库。
- 当延迟的时间满足条件之后,这些存储的数据就会被投递到真正的消息队列里面。紧接着消费者就可以消费到这个消息了。
提示:我们可以从这里得到一个启发,就是如果要实现一个延迟队列,是可以借助数据库的。
(2)那么这个插件本身也是有很多限制的,在它的官网主页里面就有说明,其中有两个最突出的限制:
- 消息在真的被投递到目标消息队列之前,是存放在接收到了这个消息的服务端本地的 Mnesia 里面。也就是说,如果这个时候还没有刷新磁盘,那么消息就会丢失;如果这个节点不可用了,那么消息也同样会丢失。
- 不支持高并发、大数据量。显然,现实中很多场景都是要在高并发大数据量场景下使用延迟消息的,比如说订单超时取消。因此这个缺点也限制了这个插件被广泛使用。
2,使用 ttl 和死信队列实现
(1)除了使用插件外,开发者也可以自己手动实现延迟消息。这就要利用到 RabbitMQ 的 ttl(Time To Live,也就是过期时间)功能和所谓的死信队列了。死信队列是一种逻辑上的概念,也就是说它本身只是一个普通的队列。而死信的意思是指过期的无法被消费的消息,这些消息会被投送到这个死信队列。
- 简单来说,就是开发者准备一个队列 delay_queue,为这个 delay_queue 设置过期时间,这个 delay_queue 不需要消费者。然后把你真实的业务 biz_queue 绑定到这个 delay_queue,作为它的死信队列。
- 生产者发送消息到 delay_queue,因为没有消费者,所以消息会过期。过期之后的消息被转发到死信队列,也就是 biz_queue 里面。这时候消费者就能拿到消息了。
![](http://hangge.com/blog_uploads/202402/2024022916483559760.png)
(2)这种方案并没有插件的那两个缺点。但是 ttl 的设置是在队列级别上,也就是说一个 delay_queue 的延时是固定的,不能做到随机。
- 比如说我这一条消息延后三分钟,另外一条消息延后五分钟,这是不可能的。因此,我们可能需要创建很多不同 ttl 的 delay_queue 才能满足业务需要。
三、Kafka 实现延迟消息方式1:利用定时任务调度
1,方案介绍
(1)利用定时任务来实现延迟消息是最好、最简单的办法。
- 对于一个延迟消息来说,一个延迟到 30 分钟后才可以被消费的消息,也可以认为是 30 分钟后才可以发送。也就是说,我们可以设定一个定时任务,这个任务会在 30 分钟后把消息发送到消息服务器上。
(2)定时任务调度最好是通过解决了持久化的分布式任务平台。那么业务发送者就相当于注册一个任务,这个任务就是在 30 分钟之后发送一条消息到 Kafka 上。之后业务消费者就能够消费了。
![](http://hangge.com/blog_uploads/202402/2024022917232815242.png)
2,方案的缺点
这个方案的昀大缺点是支撑不住高并发。这是因为绝大多数定时任务中间件都没办法支撑住高并发、大数据的定时任务调度,所以只有应用规模小,延迟消息也不多的话,才可以考虑使用这个方案。如果想要支持高并发、大数据的延迟方案,还是要考虑利用消息队列。
四、Kafka 实现延迟消息方式2:分区设置不同延迟时间
1,方案介绍
(1)该方案的基本架构图如下,这里关键的角色是 delay_topic 和延迟消费组:
- 创建了一个 delay_topic,这个topic有 N 个分区,每个分区设定了不同的延迟时间,用来接收不同延迟时间的消息。比如说在下图中分成了 p0、p1、p2 三个分区,分别用于接收延迟时间为 1min、3min 和 10min 的消息。
- 然后我们创建了一个消费组去消费这个 delay_topic,延迟消费组会创建出和分区数量一样的消费者,每一个消费者消费一个分区。每个消费者在读取到一条消息之后,就会根据消息里面的延迟时间来等待一段时间。等待完之后,再把消息发送到业务方真正的 topic 上(如下图的 biz_topic)
![](http://hangge.com/blog_uploads/202403/2024030108490180042.png)
(2)其中分区 N 是根据业务来设计,比如:
- 5 个分区:延迟时间分别是 1min、3min、5min、10min、30min。
- 10 个分区:延迟时间分别是 1min、3min,5min、10min、15min、30min、60min、90min、120min、180min。
2,rebalance 问题
(1)在这个方案中,因为消费者睡眠了,睡眠期间不会消费消息,所以 Kafka 就会判定这个消费者已经崩溃了,从而触发 rebalance。等睡眠结束之后,重新消费,就不一定还是消费原本的那个分区了。
![](http://hangge.com/blog_uploads/202403/2024030108521356733.png)
(2)为了避免这个问题,就需要确保在睡眠期间不会触发 rebalance。因此需要利用 Kafka 的暂停(Pause)功能,在睡眠结束之后,再恢复(Resume)。
提示:Kafka 的暂停消费,并不是不再发起 Poll 请求,而是 Poll 了但是不会真的拉消息,或者说相当于拉取 0 条数据。这样可以让 Kafka 始终认为消费者还活着。
(3)所以整体逻辑如下:
- 拉取一条消息,假如说 offset = N,查看剩余的延迟时间 t。
- 暂停消费,睡眠一段时间 t。
- 睡眠结束之后,恢复消费,继续从 offset = N 开始消费。
3,一致性问题
(1)从服务端拉取到消息,然后转发到 biz_topic 里面这个流程中,是先提交消息,还是先转发?
- 如果是先提交,那么就会出现消息提交了,但是还来不及转发 biz_topic 就宕机的情况,这显然不能容忍。
- 但是如果先转发 biz_topic,然后提交。那如果提交之前宕机了,后面恢复过来,又会转发一次。
(2)要解决一致性问题,必须先转发 biz_topic,然后再提交。同时还需要业务方的配合。
- 也就是说,如果在转发 biz_topic 之后,提交失败了,下一次就还可以重试,那么 biz_topic 就可能收到两条同样的消息。在这种场景下,就只能要求消费者做到幂等。
- 当然,即便不用延迟消息,消费者最好也要做到幂等的。因为发送方为了确保发送成功,本身就可能重试。
4,方案的优缺点
(1)优点:这个方案最大的优点就是足够简单,对业务方的影响很小,业务方只需要根据自己的延迟时间选择正确的分区就可以了。
(2)缺点:这个方案也有两个突出的缺点,就是延迟时间必须预先设定好、分区间负载不均匀。
- 第一个是延迟时间必须预先设定好,比如只能允许延迟 1min、3min 或者 10min 的消息,不支持随机延迟时间。在一些业务场景下,需要根据具体业务数据来计算延迟时间,那么这个就不适用了。
- 第二个是分区之间负载不均匀。比如很多业务可能只需要延迟 3min,那么 1min 和 10min 分区的数据就很少。这会进一步导致一个问题,就是负载高的分区会出现消息积压的问题。
提示:在这里,很多解决消息积压的手段都无法使用,所以只能考虑多设置几个延迟时间相近的分区,比如说在 3min 附近设置 2min30s,3min30s 这种分区来分摊压力。
五、Kafka 实现延迟消息方式3:基于MySQL转储方案
1,方案介绍
(1)该方案的实现逻辑如下:
- 创建一个 delay_topic,业务发送者把消息发送到这个 topic 里面,消息里面带上了需要延迟的时间。
- 有一个延迟消费者,它会消费 delay_topic 里面的消息,转储到数据库中。
- 还有一个延迟发送者,它会轮询数据库里的消息,把已经可以转发出去的消息转发到真正的 biz_topic 上。发送完之后,延迟发送者把数据库的状态更新成已发送。
- 最后业务消费者消费 biz_topic。
![](http://hangge.com/blog_uploads/202403/2024030109172417390.png)
(2)如何支撑住高并发是该方案要解决的一个痛点。由于此方案最明显的性能瓶颈就是 MySQL,因为这个场景是一个写密集的场景。所以要想撑住高并发就要想办法提高 MySQL 的性能。
- 当然最佳的策略还是换一个存储结构,比如说换 TiDB 或者 Elasticsearch。
- 如果一定要使用 MySQL 的话,可以采用分区表、表交替、分库分表、批量操作几个优化方案。
2,分区表
(1)最简单的优化方案就是使用分区表,因为延迟消息是一个时效性很强的数据,也就是说,我们完全可以按照发送时间,也就是延迟之后具体的发送时间点来分区。
(2)分区表根据并发量选择按月分、按周分、按天分(例如在并发不是很高的时候,可以按照周来分区。在并发很高的时候,可以按照天来分区)。历史分区就可以直接清理掉。
3,表交替
表交替方案的意思是准备两个表,然后交替写、交替查询。
- 比如准备两个表 tab_0、tab_1,今天用 tab_0,明天用 tab_1。当用 tab_1 的时候就可以直接清空(TRUNCATE)tab_0 的数据,反过来也是这样。TRUNCATE 本身很快,所以没什么性能问题。
- 这种按天交替的方案对延迟时间是有限制的,延迟时间不能超过一天。
![](http://hangge.com/blog_uploads/202403/2024030109465955549.png)
4,分库分表
(1)如果并发确实非常高,那么就只能考虑采用分库分表的方案了。这里分库分表也很简单,只需要按照 biz_topic 的名字来分库分表就可以了。而且每一张表可以叠加前面的分区表和交替表的方案,进一步提高性能。
(2)不过这个方案也有一定的隐患,第一个问题就是不同 topic 的并发度不一样,比如说 biz_topic_1 的并发只有 100,而 biz_topic_2 的并发有 10000,那么按照 biz_topic 来分,就会出现不同库不同表的压力差异很大的问题。
- 要解决这个问题,如果不考虑消息有序性的问题,那么也可以考虑轮询。比如说分库分表是 4 * 8 = 32 张表,那么就可以要求每一个延迟消费者,轮流往这些表里插入数据。因为延迟消息有一个很显著的特点,就是查找的时候只会按照发送时间来找,所以随机插入都没问题。
- 比如说我有一个消息发送给 biz_topic_1,要求是一分钟后发出去。那么不管这个消息被存在哪个表,延迟发送者都可以找出来,然后转发到biz_topic_1。
![](http://hangge.com/blog_uploads/202403/2024030111035087834.png)
5,批量操作
(1)我们还可以利用批量操作来减轻 MySQL 的压力。对于延迟消费者来说,它可以消费了一批数据之后再批量插入到数据库里面,然后再提交这一批消息。对于延迟发送者来说,当发送了一批数据之后,再批量把这些消息更新为已发送。
(2)批量操作方案类似于分区设置不同延时时间方案,会使数据一致性问题更加严重,不过只需要消费者做到幂等就可以了。因为本质上,这里的一致性问题要么是因为延迟消费者重复消费,要么就是因为延迟发送者重复发送。但不管是哪个原因,消费者幂等都可以解决问题。
全部评论(0)