基于redisson实现延迟队列

标签：mes   最大   except   pull   问题   引入   offer   async   提醒   

业务场景

最近公司的一个老项目有一个需求，需要根据后台管理员维护的时间来做一个定时任务的推送，用来推送企业微信的一些提醒消息，这个时间由于是业务人员操作，还有不确定性，其次还要受制于项目现有技术栈的限制，感觉有点难搞，还好项目在解决登录共享session的时候引入了redis，最开始的思路把维护的定时任务存储在redis然后根据redis key失效触发事件的特性来实现业务逻辑的处理，这样其实可以行得通，然后考虑到有引入redisson客户端，既然有API可以引用，那就用延迟队列来实现，既然思路有了，说干就干。

思路分析

使用redisson来做延迟队列还有两种处理方案。

• 1.直接在添加定时任务的时候就维护消息到延迟队列，但是这种情况考虑到redis挂了，数据丢失，还要配置好持久化方案，但是准确性更好，定时效果更好。
• 2.任务维护时不用处理，开启一个后台定时线程来每隔一段时间扫描定时任务，将未来的定时任务维护到延迟队列，这样由于数据存储在数据库，不用考虑redis的稳定性，但是要考虑到扫描周期带来的延迟，还有重复入队列的问题，要解决的更多。

最后权衡利弊，考虑到redis挂的可能性低和定时的准确性，就采用了第一种处理逻辑。

代码

创建redisson配置类

package com.cwn.wizbank.common;


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

import java.io.IOException;

/**
 * @author xiang xiaocheng
 * @version 1.0
 * @site chsoul.cnblogs.com
 * @date 2021/2/7 21:43
 */
@Configuration
public class RedissonConfig {
    @Bean
    public RedissonClient redissonClient() {
        Config config = null;
        try {
            config = Config.fromYAML(RedissonConfig.class.getClassLoader().getResource("redissonConfig.yaml"));
        } catch (IOException e) {
            e.printStackTrace();
        }
        return Redisson.create(config);
    }
}

redisson配置文件

singleServerConfig:

  #（连接空闲超时，单位：毫秒）
  idleConnectionTimeout: 10000
  #ping命令超时时间
  pingTimeout: 1000
  #（连接超时，单位：毫秒）
  connectTimeout: 10000
  #（命令等待超时，单位：毫秒）
  timeout: 3000
  #（命令失败重试次数）
  retryAttempts: 3
  #（命令重试发送时间间隔，单位：毫秒）
  retryInterval: 1500
  #（重新连接时间间隔，单位：毫秒）
  reconnectionTimeout: 3000
  #（执行失败最大次数）
  failedAttempts: 3
  #（密码）
  password: password
  #（单个连接最大订阅数量）
  subscriptionsPerConnection: 5
  #（客户端名称）
  clientName: clientName
  #（节点地址）
  address: "redis://ip:port"
  #（发布和订阅连接的最小空闲连接数）
  subscriptionConnectionMinimumIdleSize: 1
  #（发布和订阅连接池大小）
  subscriptionConnectionPoolSize: 10
  #（最小空闲连接数）
  connectionMinimumIdleSize: 2
  #（连接池大小）
  connectionPoolSize: 20
  #（数据库编号）
  database: 6
  #（是否启用DNS监测）
  dnsMonitoring: false
  #（DNS监测时间间隔，单位：毫秒）
  dnsMonitoringInterval: 5000
threads: 0
nettyThreads: 0
codec: !<org.redisson.codec.JsonJacksonCodec> {}
useLinuxNativeEpoll: false

创建单例延迟队列工具类

package com.cw.wizbank.util;

import com.cwn.wizbank.common.TaskDTO;
import com.cwn.wizbank.entity.Article;
import com.cwn.wizbank.utils.SpringContextUtil;
import org.apache.log4j.Logger;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * @author xiang xiaocheng
 * @version 1.0
 * @site chsoul.cnblogs.com
 * @date 2021/2/9 10:53
 */
public class DelayQueueTaskUtils {
    private RedissonClient redissonClient = (RedissonClient) SpringContextUtil.getBean("redissonClient");
    private final String qName = "QYWX-TIMER-TASK";
    private Logger logger = Logger.getLogger(DelayQueueTaskUtils.class);

    private static volatile DelayQueueTaskUtils delayQueueTaskUtils;

    private DelayQueueTaskUtils (){};
    public static DelayQueueTaskUtils getInstance(){
        if (delayQueueTaskUtils == null){
            synchronized (DelayQueueTaskUtils.class){
                if (delayQueueTaskUtils==null){
                    delayQueueTaskUtils = new DelayQueueTaskUtils();
                }
            }
        }
        return delayQueueTaskUtils;
    }

    public synchronized void pullTask(Article article) {
        RBlockingQueue<Object> blockingQueue = redissonClient.getBlockingQueue(qName);
        RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingQueue);
        TaskDTO<Article> task = new TaskDTO<>();
        task.setTaskName(article.getArt_title());
        task.setTaskBody(article);
        long currentTimeMillis = System.currentTimeMillis();
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        try {
            Date date = dateFormat.parse(article.getArt_push_msg_time());
            long pushTime = date.getTime();
            delayedQueue.offerAsync(task, pushTime - currentTimeMillis, TimeUnit.MILLISECONDS);
            logger.warn("------------> 添加任务：" + article.getArt_title());
        } catch (ParseException e) {
            logger.error("------------> 添加任务失败："+article.getArt_title());
            e.printStackTrace();
        }
    }
}

创建任务对象DTO

package com.cwn.wizbank.common;

import java.io.Serializable;

/**
 * @author xiang xiaocheng
 * @version 1.0
 * @site chsoul.cnblogs.com
 * @date 2021/2/8 9:25
 */
public class TaskDTO<T> implements Serializable {
    private String taskName;
    private T taskBody;

    public String getTaskName() {
        return taskName;
    }

    public void setTaskName(String taskName) {
        this.taskName = taskName;
    }

    public T getTaskBody() {
        return taskBody;
    }

    public void setTaskBody(T taskBody) {
        this.taskBody = taskBody;
    }
}

getInstance来获取单例队列对象，pullTask来推送任务到队列，TaskDTO封装消息实体，其中Article为具体业务类可根据业务而定。

创建队列监听服务类

package com.cwn.wizbank.services;

import com.cw.wizbank.article.ArticleModuleParam;
import com.cw.wizbank.util.QywxPushUtils;
import com.cwn.wizbank.common.TaskDTO;
import com.cwn.wizbank.entity.Article;
import org.apache.log4j.Logger;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;

/**
 * @author xiang xiaocheng
 * @version 1.0
 * @site chsoul.cnblogs.com
 * @date 2021/2/7 21:50
 */
@Service
public class QywxTimerTaskService {
    @Autowired
    private RedissonClient redissonClient;
    private final String qName = "QYWX-TIMER-TASK";
    private Logger logger = Logger.getLogger(QywxTimerTaskService.class);

    @PostConstruct
    public void execute() {
        RBlockingQueue<Object> blockingQueue = redissonClient.getBlockingQueue(qName);
        new Thread(() -> {
            while (true) {
                try {
                    TaskDTO obj = (TaskDTO) blockingQueue.take();
                    Article article = (Article) obj.getTaskBody();
                    logger.warn("--> 执行推送任务");
                    logger.warn("--> 标题：" + article.getArt_title());
                    logger.warn("--> 推送渠道：" + (0 ==article.getArt_push_msg_channel() ? "牧原大学" : "文化牧原"));
                    logger.warn("--> 推送类型：" + (0 == article.getArt_push_msg_time_type() ? "立即推送" : "定时推送"));
                    logger.warn("--> 推送时间：" + article.getArt_push_msg_time());
                    QywxPushUtils.sendMessage(article);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

关于API中队列的操作

入队列

• add,put,offer,offerAsync（异步操作）并没有太大区别，而且DelayedQueue跟添加顺序无关，保证了最快过期的对象排在head位置。

出队列

• peek：获取队列的head对象，但不是从队列中移除。如果队列空，返回空
• poll ：?获取并移出队列head对象，如果head没有超时，返回空
• take ：?获取并移出队列head对象，如果没有超时head对象，会wait当前线程知道有对象满足超时条件

总结

根据具体的业务场景寻找适合的技术路径来解决问题，目前基本已经满足了延时推送的需求，业务总线就是入队列，出队列消费。对于上面提到的第二种解决途径，入队列的时候会出现重复添加的情况，这个时候就需要做一个重复校验，但是基于原有API没有找到比较好的解决方案，这里提供一个思路，根据消息ID维护一个主键列表每次添加的时候，做重复校验，消费完成的时候再从这个列表移除任务ID，可以实现防止重复添加的情况。
入队列校验

if (set.contains(article.getArt_id())) {
    logger.warn("------------> 任务已经存在:：" + article.getArt_title());
} else {
    set.add(article.getArt_id());
    delayedQueue.offerAsync(taskDTO, pushTime - currentTimeMillis, TimeUnit.MILLISECONDS);
    logger.warn("------------> 添加任务：" + article.getArt_title());
}

移除列表ID

QywxTimerTaskSchedule.set.remove(article.getArt_id());

基于redisson实现延迟队列

标签：mes   最大   except   pull   问题   引入   offer   async   提醒   

原文地址：https://www.cnblogs.com/chsoul/p/14394744.html