Usage Scenarios

1. The order was successfully placed but not paid for 30 minutes. Payment timeout，Automatic cancellation of order

2. Order receipt，No evaluation will be conducted within 7 days after receipt. The order timed out and was not evaluated，The system defaults to a positive review

3. The order was placed successfully，The merchant did not receive the order within 5 minutes，The order was cancelled

4.The delivery timed out xff0c;Push SMS reminder

……

For scenarios with long delays and low real-time performance, we can use task scheduling to process polling at regular intervals. For example,：xxl-job

Today we adopt a relatively simple and lightweight method-using Redis' delay queue for processing. Of course there are better solutions - you can choose the optimal solution based on the company's technology selection and business system. For example, use the message middleware Kafka and RabbitMQ's delay queue

Let's not discuss its implementation principle. Let's go directly to the actual code and first implement the Redis-based delay queue

1. Introduce Redisson dependency

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.10.5</version>
</dependency>

2. Nacos configuration Redis connection

spring:
  redis:
    host: 127.0.0.1
    port: 6379
    password: 123456
    database: 12
    timeout: 3000

3. Create RedissonConfig configuration

/**
 * Created by LPB on 2020/04/20.
 */
@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);
	}

}

4. Encapsulate Redis delay queue tool class

/**
 * redis延遲隊(duì)列工具
 * Created by LPB on 2021/04/20.
 */
@Slf4j
@Component
public class RedisDelayQueueUtil {

    @Autowired
	private RedissonClient redissonClient;

    /**
     * 添加延遲隊(duì)列
     * @param value 隊(duì)列值
     * @param delay 延遲時(shí)間
     * @param timeUnit 時(shí)間單位
     * @param queueCode 隊(duì)列鍵
     * @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("(添加延時(shí)隊(duì)列成功) 隊(duì)列鍵&#xff1a;{}&#xff0c;隊(duì)列值&#xff1a;{}&#xff0c;延遲時(shí)間&#xff1a;{}", queueCode, value, timeUnit.toSeconds(delay) + "秒");
        } catch (Exception e) {
            log.error("(添加延時(shí)隊(duì)列失敗) {}", e.getMessage());
            throw new RuntimeException("(添加延時(shí)隊(duì)列失敗)");
        }
    }

	/**
	 * 獲取延遲隊(duì)列
	 * @param queueCode
	 * @param <T>
	 * @return
	 * @throws InterruptedException
	 */
    public <T> T getDelayQueue(String queueCode) throws InterruptedException {
        RBlockingDeque<Map> blockingDeque = redissonClient.getBlockingDeque(queueCode);
        T value  = (T) blockingDeque.take();
        return value;
	}
}

5. Create a delay queue business enumeration

/**
 * 延遲隊(duì)列業(yè)務(wù)枚舉
 * Created by LPB on 2021/04/20.
 */
@Getter
@NoArgsConstructor
@AllArgsConstructor
public enum RedisDelayQueueEnum {

	ORDER_PAYMENT_TIMEOUT("ORDER_PAYMENT_TIMEOUT","訂單支付超時(shí)，自動(dòng)取消訂單", "orderPaymentTimeout"),
	ORDER_TIMEOUT_NOT_EVALUATED("ORDER_TIMEOUT_NOT_EVALUATED", "訂單超時(shí)未評(píng)價(jià)，系統(tǒng)默認(rèn)好評(píng)", "orderTimeoutNotEvaluated");

	/**
	 * 延遲隊(duì)列 Redis Key
	 */
	private String code;

	/**
	 * 中文描述
	 */
	private String name;

	/**
	 * 延遲隊(duì)列具體業(yè)務(wù)實(shí)現(xiàn)的 Bean
	 * 可通過(guò) Spring 的上下文獲取
	 */
	private String beanId;

}

6. Define the delay queue executor

/**
 * 延遲隊(duì)列執(zhí)行器
 * Created by LPB on 2021/04/20.
 */
public interface RedisDelayQueueHandle<T> {

	void execute(T t);

}

7. Create the Bean， defined in the enumeration and implement the delay queue executor

• OrderPaymentTimeout：Order payment timeout delay queue processing class

/**
 * 訂單支付超時(shí)處理類(lèi)
 * Created by LPB on 2021/04/20.
 */
@Component
@Slf4j
public class OrderPaymentTimeout implements RedisDelayQueueHandle<Map> {
	@Override
	public void execute(Map map) {
		log.info("(收到訂單支付超時(shí)延遲消息) {}", map);
		// TODO 訂單支付超時(shí)&#xff0c;自動(dòng)取消訂單處理業(yè)務(wù)...

	}
}

• DelayQueueProcessorForUnevaluatedOrders: Delay queue processing class for processing unevaluated orders , used when the order times out and is not evaluated

/**
 * 訂單超時(shí)未評(píng)價(jià)處理類(lèi)
 * Created by LPB on 2021/04/20.
 */
@Component
@Slf4j
public class OrderTimeoutNotEvaluated implements RedisDelayQueueHandle<Map> {
	@Override
	public void execute(Map map) {
		log.info("(收到訂單超時(shí)未評(píng)價(jià)延遲消息) {}", map);
		// TODO 訂單超時(shí)未評(píng)價(jià)&#xff0c;系統(tǒng)默認(rèn)好評(píng)處理業(yè)務(wù)...

	}
}

8. Create a delay queue consumption thread，Start after the project is started

/**
 * 啟動(dòng)延遲隊(duì)列
 * Created by LPB on 2021/04/20.
 */
@Slf4j
@Component
public class RedisDelayQueueRunner implements CommandLineRunner {

	@Autowired
	private RedisDelayQueueUtil redisDelayQueueUtil;

	@Override
	public void run(String... args) {
		new Thread(() -> {
			while (true){
				try {
					RedisDelayQueueEnum[] queueEnums = RedisDelayQueueEnum.values();
					for (RedisDelayQueueEnum queueEnum : queueEnums) {
						Object value = redisDelayQueueUtil.getDelayQueue(queueEnum.getCode());
						if (value != null) {
							RedisDelayQueueHandle redisDelayQueueHandle = SpringUtil.getBean(queueEnum.getBeanId());
							redisDelayQueueHandle.execute(value);
						}
					}
				} catch (InterruptedException e) {
					log.error("(Redis延遲隊(duì)列異常中斷) {}", e.getMessage());
				}
			}
		}).start();
		log.info("(Redis延遲隊(duì)列啟動(dòng)成功)");
	}
}

The above steps ，Redis delay queue core code has been completed，Let’s write a test interface，Use PostMan to simulate and test it

9. Create a test interface，Simulation Add a delay queue

/**
 * 延遲隊(duì)列測(cè)試
 * Created by LPB on 2020/04/20.
 */
@RestController
public class RedisDelayQueueController {

	@Autowired
	private RedisDelayQueueUtil redisDelayQueueUtil;

	@PostMapping("/addQueue")
	public void addQueue() {
		Map<String, String> map1 = new HashMap<>();
		map1.put("orderId", "100");
		map1.put("remark", "訂單支付超時(shí)&#xff0c;自動(dòng)取消訂單");

		Map<String, String> map2 = new HashMap<>();
		map2.put("orderId", "200");
		map2.put("remark", "訂單超時(shí)未評(píng)價(jià)&#xff0c;系統(tǒng)默認(rèn)好評(píng)");

		// 添加訂單支付超時(shí)&#xff0c;自動(dòng)取消訂單延遲隊(duì)列。為了測(cè)試效果&#xff0c;延遲10秒鐘
		redisDelayQueueUtil.addDelayQueue(map1, 10, TimeUnit.SECONDS, RedisDelayQueueEnum.ORDER_PAYMENT_TIMEOUT.getCode());

		// 訂單超時(shí)未評(píng)價(jià)&#xff0c;系統(tǒng)默認(rèn)好評(píng)。為了測(cè)試效果&#xff0c;延遲20秒鐘
		redisDelayQueueUtil.addDelayQueue(map2, 20, TimeUnit.SECONDS, RedisDelayQueueEnum.ORDER_TIMEOUT_NOT_EVALUATED.getCode());
	}

}

10. Start the SpringBoot project，Use the PostMan calling interface to add a delay queue

• You can see through the Redis client that the two delay queues have been Added successfully

• Check the IDEA console log to see that the delay queue has been consumed successfully

The above is the detailed content of How SpringBoot integrates Redisson to implement delay queue. For more information, please follow other related articles on the PHP Chinese website!