redis使用延迟队列_redisdelayqueueenum

pom文件

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
		</dependency>
		
		<dependency>
			<groupId>org.projectlombok</groupId>
			<artifactId>lombok</artifactId>
			<optional>true</optional>
		</dependency>
		
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
		
		<dependency>
			<groupId>org.redisson</groupId>
			<artifactId>redisson-spring-boot-starter</artifactId>
			<version>3.10.5</version>
		</dependency>

		<dependency>
			<groupId>cn.hutool</groupId>
			<artifactId>hutool-all</artifactId>
			<version>5.8.15</version>
		</dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

yml文件：

server:
  port: 8888

spring:
  redis:
    host: 127.0.0.1
    port: 6379
    password: 123
    database: 12
1
2
3
4
5
6
7
8
9

RedissonConfig：

package com.inesa.shwater.redis.lock.config;

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RedissonConfig {
 
	@Value("${spring.redis.host}")
	private String host;
	@Value("${spring.redis.port}")
	private int port;
	@Value("${spring.redis.database}")
	private int database;
	@Value("${spring.redis.password}")
	private String password;
 
	@Bean
	public RedissonClient redissonClient() {
		Config config = new Config();
		config.useSingleServer()
			.setAddress("redis://" + host + ":" + port)
			.setDatabase(database)
			.setPassword(password);
		return Redisson.create(config);
	}
 
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

RedisDelayQueueEnum：

package com.inesa.shwater.redis.lock.config;

import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;

/**
 * 延迟队列业务枚举
 */
@Getter
@NoArgsConstructor
@AllArgsConstructor
public enum RedisDelayQueueEnum {
 
	ORDER_PAYMENT_TIMEOUT("ORDER_PAYMENT_TIMEOUT","订单支付超时，自动取消订单", "orderPaymentTimeout"),
	ORDER_TIMEOUT_NOT_EVALUATED("ORDER_TIMEOUT_NOT_EVALUATED", "订单超时未评价，系统默认好评", "orderTimeoutNotEvaluated");
 
	/**
	 * 延迟队列 Redis Key
	 */
	private String code;
 
	/**
	 * 中文描述
	 */
	private String name;
 
	/**
	 * 延迟队列具体业务实现的 Bean
	 * 可通过 Spring 的上下文获取
	 */
	private String beanId;
 
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

RedisDelayQueueRunner：

package com.inesa.shwater.redis.lock.config;

import cn.hutool.extra.spring.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @Description: 启动延迟队列 （如果是微服务，避免被其他服务消费，应该把此RUNNER和ENUM提取到其他相关delay的上一级）
 */
@Slf4j
@Component
public class RedisDelayQueueRunner implements CommandLineRunner {

    @Autowired
    private RedissonClient redissonClient ;

    @Override
    public void run(String... args) {
        RedisDelayQueueEnum[] queueEnums = RedisDelayQueueEnum.values();
        for (RedisDelayQueueEnum queueEnum : queueEnums) {
            RBlockingDeque<Map> blockingDeque = redissonClient.getBlockingDeque(queueEnum.getCode());
            //避免消息伪丢失（应用重启未消费），官网推荐
            redissonClient.getDelayedQueue(blockingDeque) ;
            new Thread(() -> {
                while (true) {
                    try {
                        Object take = blockingDeque.take();
                        if(take!=null) {
                            RedisDelayQueueHandle redisDelayQueueHandle = SpringUtil.getBean(queueEnum.getBeanId());
                            redisDelayQueueHandle.execute(take);
                        }
                    } catch (InterruptedException e) {
                        log.error("(Redis延迟队列异常中断) {}", e.getMessage());
                    }
                }
            }).start();
        }
        log.info("Redis延迟队列启动成功");
    }
}


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

RedisDelayQueueHandle：

package com.inesa.shwater.redis.lock.config;

/**
 * 延迟队列执行器
 */
public interface RedisDelayQueueHandle<T> {
 
	void execute(T t);
 
}
1
2
3
4
5
6
7
8
9
10

RedisDelayQueueUtil：

package com.inesa.shwater.redis.lock.config;

import io.micrometer.core.instrument.util.StringUtils;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * redis延迟队列工具
 * Created by LPB on 2021/04/20.
 */
@Slf4j
@Component
public class RedisDelayQueueUtil {
 
    @Autowired
	private RedissonClient redissonClient;
 
    /**
     * 添加延迟队列
     * @param value 队列值
     * @param delay 延迟时间
     * @param timeUnit 时间单位
     * @param queueCode 队列键
     * @param <T>
     */
    public <T> void addDelayQueue(T value, long delay, TimeUnit timeUnit, String queueCode){
        try {
            RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueCode);
            RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
            delayedQueue.offer(value, delay, timeUnit);
			log.info("(添加延时队列成功) 队列键：{}，队列值：{}，延迟时间：{}", queueCode, value, timeUnit.toSeconds(delay) + "秒");
        } catch (Exception e) {
            log.error("(添加延时队列失败) {}", e.getMessage());
            throw new RuntimeException("(添加延时队列失败)");
        }
    }
 
	/**
	 * 获取延迟队列
	 * @param queueCode
	 * @param <T>
	 * @return
	 * @throws InterruptedException
	 */
    public <T> T getDelayQueue(String queueCode) throws InterruptedException {
        RBlockingDeque<Map> blockingDeque = redissonClient.getBlockingDeque(queueCode);
        //避免消息伪丢失（应用重启未消费），官网推荐
        redissonClient.getDelayedQueue(blockingDeque) ;
        T value  = (T) blockingDeque.take();
        return value;
	}


    public boolean removeDelayedQueue(@NonNull Object o, @NonNull String queueCode) {
        if (StringUtils.isBlank(queueCode) || Objects.isNull(o)) {
            return false;
        }
        try {
            RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueCode);
            RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
            boolean flag = delayedQueue.remove(o);
            if(flag){
                log.info("(删除延时队列保证唯一性) 队列键：{}，队列值：{}", queueCode, o);
            }
            delayedQueue.destroy();
            return flag;
        } catch (Exception e) {
            log.error("(删除延时队列异常) 队列键：{}，队列值：{},错误信息：{}",queueCode, o, e.getMessage());
            throw new RuntimeException("(删除延时队列异常)");
        }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

OrderPaymentTimeout：

package com.inesa.shwater.redis.lock.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 订单支付超时处理类
 */
@Component
@Slf4j
public class OrderPaymentTimeout implements RedisDelayQueueHandle<Map> {
	@Override
	public void execute(Map map) {
		log.info("(收到订单支付超时延迟消息) {}", map);
		// TODO 订单支付超时，自动取消订单处理业务...
 
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

OrderTimeoutNotEvaluated：

package com.inesa.shwater.redis.lock.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 订单超时未评价
 */
@Component
@Slf4j
public class OrderTimeoutNotEvaluated implements RedisDelayQueueHandle<Map> {
    @Override
    public void execute(Map map) {
        log.info("(收到订单超时未评价，系统默认好评) {}", map);
        // TODO 订单超时未评价，系统默认好评，订单处理业务...

    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

RedisDelayQueueController：

package com.inesa.shwater.redis.lock.controller;

import com.inesa.shwater.redis.lock.config.RedisDelayQueueEnum;
import com.inesa.shwater.redis.lock.config.RedisDelayQueueUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * 延迟队列测试
 */
@RestController
public class RedisDelayQueueController {
 
	@Autowired
	private RedisDelayQueueUtil redisDelayQueueUtil;
 
	//@PostMapping("/addQueue")
	@GetMapping("/addQueue")
	public void addQueue() {
		Map<String, String> map1 = new HashMap<>();
		map1.put("orderId", "100");
		map1.put("remark", "订单支付超时，自动取消订单");
 
		Map<String, String> map2 = new HashMap<>();
		map2.put("orderId", "200");
		map2.put("remark", "订单超时未评价，系统默认好评");
 
		// 添加订单支付超时，自动取消订单延迟队列。为了测试效果，延迟10秒钟
		redisDelayQueueUtil.addDelayQueue(map1, 10, TimeUnit.SECONDS, RedisDelayQueueEnum.ORDER_PAYMENT_TIMEOUT.getCode());
 
		// 订单超时未评价，系统默认好评。为了测试效果，延迟20秒钟
		redisDelayQueueUtil.addDelayQueue(map2, 20, TimeUnit.SECONDS, RedisDelayQueueEnum.ORDER_TIMEOUT_NOT_EVALUATED.getCode());
	}
 
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41