• springboot+redis实现延迟队列（内含Redisson版本）

    一：利用redis的zset实现消息队列

    使用场景
    1、下单成功，30分钟未支付。支付超时，自动取消订单

    2、订单签收，签收后7天未进行评价。订单超时未评价，系统默认好评

    3、下单成功，商家5分钟未接单，订单取消

    4、配送超时，推送短信提醒

    ......

    对于延时比较长的场景、实时性不高的场景，我们可以采用任务调度的方式定时轮询处理。如：xxl-job

    今天我们采用一种比较简单、轻量级的方式，使用 Redis 的延迟队列来进行处理。当然有更好的解决方案，可根据公司的技术选型和业务体系选择最优方案。如：使用消息中间件Kafka、RabbitMQ 的延迟队列

    原理：开启一个守护线程对消息进行轮询消费，利用zset的有序性，ZSet 有一个 Score 属性可以用来存储延迟执行的时间,将到达过期时间的消息去除，不能的实现类处理不同的业务。

    1：延迟消息实体类

    1. import com.yx.constant.IdWorker;
    2.  
    3. import java.io.Serializable;
    4.  
    5. /**
    6.  * 延迟消息实体类
    7.  */
    8. public class DelayMessage implements Serializable {
    9.     /**
    10.      * 队列id 用原子类生产
    11.      */
    12.     private String seqId;
    13.     /**
    14.      * 消息主体
    15.      */
    16.     private Object body;
    17.     /**
    18.      * 消息消费的时间戳
    19.      */
    20.     private long expire;
    21.  
    22.     public DelayMessage(){}
    23.  
    24.     public DelayMessage(Object body, long expire) {
    25.         this.seqId = IdWorker.getNextIdStr();
    26.         this.body = body;
    27.         this.expire = expire;
    28.     }
    29.  
    30.     public DelayMessage(Object body, int delay) {
    31.        this(body, System.currentTimeMillis() + delay);
    32.     }
    33.  
    34.     public String getSeqId() {
    35.         return seqId;
    36.     }
    37.  
    38.     public Object getBody() {
    39.         return body;
    40.     }
    41.  
    42.     public long getExpire() {
    43.         return expire;
    44.     }

    2：延迟队列类

    1. import org.springframework.beans.factory.annotation.Autowired;
    2. import org.springframework.data.redis.core.RedisTemplate;
    3. import org.springframework.data.redis.core.ZSetOperations;
    4. import org.springframework.stereotype.Component;
    5.  
    6. import java.lang.reflect.Array;
    7. import java.util.Set;
    8.  
    9. /**
    10.  * 延迟队列类
    11.  */
    12. @Component
    13. public class DelayQueue {
    14.     @Autowired
    15.     private RedisTemplate redisTemplate;
    16.  
    17.     /**
    18.      * 加入一个消息
    19.      * @param queueName 队列名称
    20.      * @param message 延迟消息类
    21.      */
    22.     public void add(String queueName, DelayMessage message) {
    23.         redisTemplate.opsForZSet().add(queueName, message, message.getExpire());
    24.         synchronized (this) {
    25.             //加入一个消息后唤醒所有睡眠的线程
    26.             this.notifyAll();
    27.         }
    28.     }
    29.  
    30.     /**
    31.      * 取出一个消息
    32.      * @param queueName 队列名称
    33.      * @return
    34.      * @throws InterruptedException
    35.      */
    36.     public DelayMessage take(String queueName) throws InterruptedException {
    37.         while (true) {
    38.             ZSetOperations.TypedTuple<DelayMessage> tuple = getFirst(queueName);
    39.  
    40.             if (tuple == null) {
    41.                 synchronized (this) {
    42.                     //如果该队列为空则休眠该线程
    43.                     this.wait();
    44.                     tuple = getFirst(queueName);
    45.                 }
    46.             }
    47.             if (tuple != null){
    48.                 //消息到期时间戳
    49.                 long expire = tuple.getScore().longValue();
    50.                 //延迟时间
    51.                 long delay = expire - System.currentTimeMillis();
    52.                 //未到达消费时间
    53.                 if (delay > 0) {
    54.                     synchronized (this) {
    55.                         //让线程休眠delay毫秒
    56.                         this.wait(delay);
    57.                     }
    58.                 }
    59.                 /*
    60.                     这里判断消息消费时间是否大于当前时间的原因：
    61.                     当该线程执行上一步 this.wait(delay);
    62.                     正好有消息消费在休眠准备消费时；
    63.                     有消息加入执行了add 加入消息，此时唤醒了休眠的线程
    64.                     所有这里再做一次判断，防止未到时间消息就消费了
    65.                 */
    66.                 if (System.currentTimeMillis() >= expire) {
    67.                     //取出消息
    68.                     DelayMessage delayMessage = tuple.getValue();
    69.                     //把消息 id取出 建立一个key
    70.                     String lockKey = queueName + "_" + delayMessage.getSeqId() + "_lock";
    71.                     //setIfAbsent的作用是：
    72.                     //如果该key有值1，则返回false
    73.                     //如果该key没有值为1,则返回ture而且再为key加上值为1
    74.                     Boolean ok = redisTemplate.opsForValue().setIfAbsent(lockKey, "1");
    75.                     if (ok) {
    76.                         //移除redis之前存的消息
    77.                         redisTemplate.opsForZSet().remove(queueName, delayMessage);
    78.                         //删除刚刚为了保证原子性的key
    79.                         redisTemplate.delete(lockKey);
    80.                         //返回消息主体
    81.                         return delayMessage;
    82.                     }
    83.                 }
    84.             }
    85.         }
    86.     }
    87.  
    88.     /**
    89.      * 得到该队列的第一条数据
    90.      * @param queueName 队列名称
    91.      * @return
    92.      */
    93.     private ZSetOperations.TypedTuple getFirst(String queueName) {
    94.         //获取队列名称为 queueName的所有的消息
    95.         Set<ZSetOperations.TypedTuple<DelayMessage>> tuples = redisTemplate.opsForZSet().rangeWithScores(queueName, 0, -1);
    96.         if (tuples.size() == 0) {
    97.             return null;
    98.         }
    99.         ZSetOperations.TypedTuple[] array = (ZSetOperations.TypedTuple[]) Array.newInstance(ZSetOperations.TypedTuple.class, tuples.size());
    100.         return tuples.toArray(array)[0];
    101.     }

    3：延迟消息消费类 开启一个线程进行消息的消费

    1. import org.springframework.beans.factory.annotation.Autowired;
    2. import org.springframework.boot.ApplicationArguments;
    3. import org.springframework.boot.ApplicationRunner;
    4.  
    5. /**
    6.  * 延迟消息消费类 开启一个线程进行消息的消费
    7.  */
    8. public abstract class DelayQueueConsumer implements ApplicationRunner {
    9.     @Autowired
    10.     private DelayQueue queue;
    11.  
    12.     public abstract String getQueueName();
    13.  
    14.     public abstract void consume(DelayMessage message);
    15.  
    16.     public abstract void catchInterruptedException(InterruptedException e);
    17.  
    18.     @Override
    19.     public void run(ApplicationArguments args) throws Exception {
    20.         //获取队列名称
    21.         String queueName = getQueueName();
    22.         if (queueName == null) {
    23.             throw new NullPointerException("the return value of getQueueName method cannot be null");
    24.         }
    25.         Thread thread = new Thread(()->{
    26.             while (true) {
    27.                 try {
    28.                     //获取消息
    29.                     DelayMessage message = queue.take(queueName);
    30.                     //消费消息
    31.                     this.consume(message);
    32.                 } catch (Exception e) {
    33.                     e.printStackTrace();
    34.                 }
    35.             }
    36.         });
    37.         thread.setDaemon(true);
    38.         thread.start();
    39.     }

    4：示例

    1. @Autowired
    2. private DelayQueue queue;
    3.  
    4. @GetMapping("add")
    5.     public String addQueue(@RequestParam Map<String, Object> params){
    6.         long millis = System.currentTimeMillis() + 30000;
    7.         DelayMessage delayMessage = new DelayMessage(params, millis);
    8.         queue.add("remind",delayMessage);
    9.         SimpleDateFormat format = new SimpleDateFormat("hh:mm:ss");
    10.         String time = format.format(new Date(millis));
    11.         return time;
    12.     }
    1. import java.text.SimpleDateFormat;
    2. import java.util.Map;
    3.  
    4. /**
    5.  * @ClassName MessageTask
    6.  * @Author yangxi
    7.  * @Date 2019/9/8 10:42
    8.  * @Description desc
    9.  **/
    10. @Component
    11. public class MessageTask extends DelayQueueConsumer {
    12.     @Override
    13.     public String getQueueName() {
    14.         return "remind";
    15.     }
    16.  
    17.     @Override
    18.     public void consume(DelayMessage message) {
    19.         Map body = (Map) message.getBody();
    20.         String name = body.get("name").toString();
    21.         String age = body.get("age").toString();
    22.         SimpleDateFormat format = new SimpleDateFormat("hh:mm:ss");
    23.         String newTime = format.format(System.currentTimeMillis());
    24.         System.out.println("当前时间"+ newTime + age + "|||  -.-  |||" + name);
    25.     }
    26.  
    27.     @Override
    28.     public void catchInterruptedException(InterruptedException e) {
    29.  
    30.     }

    二：利用redisson的RBlockingQueue实现消息队列

    1：基于redisson实现的延时消息管理器

    1. import cn.hutool.core.thread.ThreadUtil;
    2. import cn.hutool.extra.spring.SpringUtil;
    3. import com.yx.handler.DelayMessageHandler;
    4. import com.yx.message.DelayMessage;
    5. import com.yx.message.DelayMessageType;
    6. import lombok.extern.slf4j.Slf4j;
    7. import org.redisson.Redisson;
    8. import org.redisson.api.RBlockingQueue;
    9. import org.redisson.api.RDelayedQueue;
    10. import org.redisson.api.RedissonClient;
    11. import org.springframework.stereotype.Component;
    12.  
    13. import java.util.Arrays;
    14. import java.util.Map;
    15. import java.util.concurrent.ConcurrentHashMap;
    16.  
    17. /**
    18.  * @description 基于redisson实现的延时消息管理器
    19.  */
    20. @Component
    21. @Slf4j
    22. public class RedissonDelayMessageManager implements DelayMessageManager {
    23.  
    24.     private RedissonClient redissonClient = Redisson.create();
    25.  
    26.     private RBlockingQueue<DelayMessage> rBlockingQueue = redissonClient.getBlockingDeque("redisson-delay-message-queue");
    27.  
    28.     private RDelayedQueue<DelayMessage> rDelayedQueue = redissonClient.getDelayedQueue(rBlockingQueue);
    29.  
    30.     private final Map<DelayMessageType, DelayMessageHandler> handlerMap = new ConcurrentHashMap<>(16);
    31.  
    32.     @Override
    33.     public void add(DelayMessage message) {
    34.         if (rDelayedQueue.contains(message)) {
    35.             return;
    36.         }
    37.         log.info("redisson-delay-message-queue,add message = {}", message);
    38.         rDelayedQueue.offer(message, message.getProperties().getExpire(), message.getProperties().getTimeUnit());
    39.     }
    40.  
    41.     @Override
    42.     public boolean remove(DelayMessage message) {
    43.         return rDelayedQueue.remove(message);
    44.     }
    45.  
    46.     @Override
    47.     public void destroy() {
    48.         rDelayedQueue.destroy();
    49.     }
    50.  
    51.     @Override
    52.     public void afterPropertiesSet() {
    53.         Arrays.stream(DelayMessageType.values()).forEach(delayMessageType -> handlerMap.put(delayMessageType, SpringUtil.getBean(delayMessageType.getHandler())));
    54.         Thread thread = new Thread(() -> {
    55.             while (true) {
    56.                 try {
    57.                     DelayMessage delayMessage = rBlockingQueue.take();
    58.                     log.info("redisson-delay-message-queue,consume message = {}", delayMessage);
    59.                     handlerMap.get(delayMessage.getType()).handle(delayMessage);
    60.                 } catch (InterruptedException e) {
    61.                     e.printStackTrace();
    62.                     ThreadUtil.safeSleep(1000);
    63.                     afterPropertiesSet();
    64.                 }
    65.             }
    66.         });
    67.         thread.setDaemon(true);
    68.         thread.start();
    69.     }

    2：延时消息处理器接口

    1. /**
    2.  * @description 延时消息处理器接口
    3.  */
    4. public interface DelayMessageHandler {
    5.  
    6.     /**
    7.      * 处理消息
    8.      *
    9.      * @param message 消息
    10.      */
    11.     void handle(DelayMessage message);
    12. }

    3：示例

    1. @Resource(type = RedissonDelayMessageManager.class)
    2.     private RedissonDelayMessageManager delayMessageManager;
    3.  
    4.     @GetMapping("add")
    5.     private String add() {
    6.         DelayMessage delayMessage1 = new DelayMessage();
    7.         QrCode qrCode1 = new QrCode();
    8.         qrCode1.setConfigId("123");
    9.         qrCode1.setUrl("http://www.baidu.com");
    10.         delayMessage1.setBody(JSONUtil.toJsonStr(qrCode1));
    11.         delayMessage1.setType(DelayMessageType.DELETE_QR_CODE);
    12.  
    13.         DelayMessage.DelayMessageProperties properties1 = new DelayMessage.DelayMessageProperties();
    14.         properties1.setExpire(1239);
    15.         properties1.setTimeUnit(TimeUnit.MILLISECONDS);
    16.         delayMessage1.setProperties(properties1);
    17.  
    18.         delayMessageManager.add(delayMessage1);
    19.         return "success";
    20.     }
    1. /**
    2.  * @description 任务消息处理器
    3.  */
    4. @Component
    5. @Slf4j
    6. public class ExecuteTaskDelayMessageHandler implements DelayMessageHandler {
    7.  
    8.     @Override
    9.     public void handle(DelayMessage message) {
    10.         log.info("任务延时消息处理中,message={}", message);
    11.         System.out.println(DateUtil.now() + "" + JSONUtil.toJsonStr(message));
    12.     }
    13. }

    三：项目demo地址

     delay-message-redis-demo

  • 相关阅读:
    辅助驾驶功能开发-功能规范篇(16)-2-领航辅助系统NAP-功能ODD定义
    IaC实战指南：DevOps的自动化基石
    C 语言的标识符，保留标识符，关键字
    一个纯Python构建的Web应用框架
    【JVM】类的生命周期
    记一次 HTTPS 抓包分析和 SNI 的思考
    硬件新问答
    js异步编程-题目29 实现异步任务执行器 AsyncWorker
    2023年内网穿透常用的几个工具
    Opencv图像边缘检测——Roberts算子（手写）、Sobel算子（手写和调包）、Scharr算子、Laplacian算子
  • 原文地址：https://blog.csdn.net/qq_35493807/article/details/125618796