# delay-redisson-spring-boot-starter

**Repository Path**: listen_w/delay-redisson-spring-boot-starter

## Basic Information

- **Project Name**: delay-redisson-spring-boot-starter
- **Description**: 基于 redisson 和 kafka 实现一个延时队列
- **Primary Language**: Java
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 12
- **Forks**: 16
- **Created**: 2022-09-19
- **Last Updated**: 2024-10-23

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# mongo-tool


#### 介绍
基于 redisson 和 kafka 实现一个延时队列


#### 背景
其实在工作中延迟任务的场景还是比较常见的，比如电商场景中：
- 用户下单30分钟未支付自动取消订单；
- 订单申请退款7天内未处理自动完成退款；
- 订单完成7天内自动结算；
- ....

我们最近业务中需要大量使用这种延时任务，因此不得不寻找延时任务的方案，虽然说网上延时任务的方案很多，但我感觉真正落地的而且比较简单的方案并不多见。

java延迟任务方案一般包括以下：
1. JDK 线程池或者Timer定时器；
2. DelayQueue，JDK 自带的延时队列；
3. 基于netty时间轮算法实现；
4. 基于 redis 的 zset 数据结构实现；
5. 基于消息队列 RabbitMQ 实现方案；
6. ....

#### 延迟任务方案选择
在选择延迟任务方案时，排除了JDK 自带的方案，这些都是基于应用内存的，如果你的应用发个版本啥的，延迟消息就丢失，另外，对于消息队列是可以优先选择的，但是我们使用的消息队列是 Kafka ，而 Kafka本身是不具备延迟任务功能的，而且我们也不可能因为一个延迟任务而引入新的 RabbitMQ 消息中间件，因此，我把目光放在 redis 上面。

#### 网上 kafka 延迟队列方案
其实在使用 redis 做延迟任务方案之前，在网上找到一个使用 kafka 设计延迟任务的方案
> 网上kafka实现延迟任务方案地址：https://github.com/Dewey-Ding/spring-learn

先来说说为啥我放弃了这种方案：

其实主要是这种方案满足不了我们现在的业务场景，先看以下代码：
![image.png](https://p3-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/d6f0df8aafcf4e11b4fbea59a6fc268d~tplv-k3u1fbpfcp-watermark.image?)

如图，对于使用 kafka 延迟方案，发送消息时和普通发 kafka 消息一样，只是在接收kafka消息时，使用了自定义注解，该注解有个属性：`delayTimeSec`（延迟多少时间），那就是说，这种延迟是固定的（固定在代码写死的），而我们的需求是延迟是可变化。（比如订单下单超时可以是30分钟，运营也可以调整为一个小时）

如果只是这点问题的话，那我在这个方案的基础上改一改源码，因为也能实现延迟可以变化，但是看了他的实现逻辑之后，我发现即使能实现延迟可以变化也不适合我这种场景。

这个方式中，kafka 延迟消息是通过 `KafkaDelayConsumerThread`消费者线程来实现的。具体逻辑是：
1. 线程启动时订阅指定 topic，然后开始消费消息；
2. 当 kafka 队列中无可消费的延迟消息时，消费者进入 pause 状态，不再拉取指定队列的消息；
3. 等到 pauseTime 之后再回复消费。消息的延迟时间以及 pause 时间则是从注解参数中获取。

所以，我觉得这种方案会出现以下问题（具体我没验证过，只是看代码猜测会出现这样的问题）

![image.png](https://p3-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/28685f69eb1d4f6a87c5141aa70a12fc~tplv-k3u1fbpfcp-watermark.image?)

#### 软件架构
既然网上 kafka 延迟队列方案不满足我们业务场景，那只能自己搞一个延迟任务了。

由于 redis 基本上是每个项目必备的技术方案了，所以我把 延迟任务方案放在 redis上，其实 redis 有两种方案可以实现延迟任务：
- 基于 redis 键过期事件通知方案：  
    简单来说就是 redis 如果一个键过期了是可以通知到我们的（当然需要在redis上配置一下），我们接收到过期事件后处理自己的延时业务，比如：我创建一个订单时，把订单号作为键和值并设置过期时间是30分钟写入 redis ,那么这个key过期时会有回调通知我们服务，我们就可以拿到订单号（注意过期时只能拿到键）去检查是否支付了.....
    但是这种方案也有缺点，就是它的过期通知是不稳定的（也就是说有可能我们不会收到redis的回调），而且如果redis的key很多时，会有延迟。

- 基于 redis 的 zset 数据结构  
    这种方案的原理网上也有很多就不啰嗦了

其实 redis 的客户端 Redisson 就实现了延时队列的方案，而且 API 很简单，不过我在 Redisson 的基础上，增加了 Kafka 组件，先来看看我的设计方案：

![image.png](https://p6-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/210dc8ca68ab4c4fabf8d07e93a47abc~tplv-k3u1fbpfcp-watermark.image?)

流程大概就是：业务使用方的消息发送者（比如订单创建后发送一个消息，订单超时30分钟未支付），消息发送者发送一个延迟消息到redis中，我们的延迟队列中间件启动消费者去redis消费过期消息，然后将消息重新投递到 kafka中，业务方消费者通过 Kafka 消费延迟消息。


整体概论
![image.png](https://p9-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/5fefe25b0119405d87752152a5187e02~tplv-k3u1fbpfcp-watermark.image?)

延迟消息发送接收流程：
![image.png](https://p6-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/d82df79162e14a6bbc8394318609c4b0~tplv-k3u1fbpfcp-watermark.image?)

疑问：
1. 为什么要接入 Kafka ？ 延迟队列中间件的消费者消费到的消息就是延迟消息了直接给业务方消费不就行了？

对于这个问题，我后面再结合代码解答。


#### 使用说明
那怎么使用这个延迟队列呢？我将这个工具封装成一个 starter ，总结一下使用步骤：

1. 引入依赖（项目使用 Maven 构建）
    ```
    <dependency>
        <groupId>org.delay.redisson</groupId>
        <artifactId>delay-redisson-spring-boot-starter</artifactId>
        <version>1.0.0-SNAPSHOT</version>
    </dependency>
    ```

2. 配置文件增加配置：
    
    ```yaml
    delay:
      redisson:
          enable: true      --- 打开延迟功能
          registerService: order    --- 注册服务，默认 other ，topic 存储在redis时分服务存储
          consumerSleep: 500        --- 消费者轮询时睡眠时长，默认 500毫秒  
    ```

3. 使用 `DelayQueueMessageProducer` 发延迟消息 

![image.png](https://p3-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/b9519a5d6c2643b0a2b19a9dc577f327~tplv-k3u1fbpfcp-watermark.image?)
![image.png](https://p9-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/fe3c525109fa41daa4ed07af56bb2cbd~tplv-k3u1fbpfcp-watermark.image?)

4. 使用 Kafka 注解消费消息

![image.png](https://p9-juejin.byteimg.com/tos-cn-i-k3u1fbpfcp/1bb1ae6f81ef429391a9e55cec7ad3bf~tplv-k3u1fbpfcp-watermark.image?)

以上4步就可以啦

> 要注意的地方是，你需要有 redis 和 kafka的环境，也就是说你 spring 环境里需要有 `StringRedisTemplate`、`RedissonClient`、`KafkaTemplate<Long, String>` 这几个Bean

#### 文档说明
https://juejin.cn/post/7144969196542099469