redisson简介

Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。Redisson的宗旨是促进使用者对Redis的关注分离（Separation of Concern），从而让使用者能够将精力更集中地放在处理业务逻辑上，底层采用Netty实现。

这里直接摘抄了官方介绍，有关redisson结构以及使用等更加详细的介绍大家可以去redisson官网或者git上了解学习。

git仓库地址：github.com/redisson/re…

原理解析

redisson实现延时队列其实是用TimerTask来完成，其中会涉及很多数据的写操作，，redis均是通过lua脚本保证的，下面我们来简单介绍一下流程。

redisson首先会在内存中生成三个变量，queueName、channelName、timeoutSetName，这三个变量分别以redisson_delay_queue、redisson_queue_channel、redisson_delay_queue_timeout为前缀。

当客户端提交任务时，通过lua脚本会执行以下动作：

会将提交的任务数据转为二进制数据，
将延时时间和二进制数据放入有序集合中（timeoutSetName），以（延时时间+当前时间）为分值
将二进制数据放入queueName队列中
将timeout时间（延时时间+当前时间）publish到channelName通道中

具体lua脚本如下

local value = struct.pack('dLc0', tonumber(ARGV[2]), string.len(ARGV[3]), ARGV[3]);
redis.call('zadd', KEYS[2], ARGV[1], value);
redis.call('rpush', KEYS[3], value);
// if new object added to queue head when publish its startTime 
// to all scheduler workers 
local v = redis.call('zrange', KEYS[2], 0, 0); 
if v[1] == value then 
   redis.call('publish', KEYS[4], ARGV[1]); 
end;
复制代码

之后监听channelName通道的listener会执行scheduleTask方法，该方法的具体流程如下：

将发布的timeout时间减去当前时间，获取延时时间delayTime，
如果delayTime小于10毫秒，则立即执行pushTask方法，该方法最终会调用pushTaskAsync方法，pushTaskAsync方法执行流程如下：

1>. 将有序集合timeoutSetName中的分值在（0，当前时间戳）范围的前100元素取出得到一个集合

2>. 如果集合中有值，则遍历该集合，并将元素放入blockingQueue中，同时将queueName队列中对应的元素删除。遍历完成后将timeoutSetName集合中的元素删除

如果delayTime大于10毫秒，则生成一个TimerTask任务延时delayTime后执行pushTask方法。

具体代码如下：

scheduleTask代码

private void scheduleTask(final Long startTime) {
    TimeoutTask oldTimeout = lastTimeout.get();
    if (startTime == null) {
        return;
    }
    
    if (oldTimeout != null) {
        oldTimeout.getTask().cancel();
    }
    
    long delay = startTime - System.currentTimeMillis();
    if (delay > 10) {
        Timeout timeout = connectionManager.newTimeout(new TimerTask() {                    
            @Override
            public void run(Timeout timeout) throws Exception {
                pushTask();
                
                TimeoutTask currentTimeout = lastTimeout.get();
                if (currentTimeout.getTask() == timeout) {
                    lastTimeout.compareAndSet(currentTimeout, null);
                }
            }
        }, delay, TimeUnit.MILLISECONDS);
        if (!lastTimeout.compareAndSet(oldTimeout, new TimeoutTask(startTime, timeout))) {
            timeout.cancel();
        }
    } else {
        pushTask();
    }
}
复制代码

pushTaskAsync方法流程则是通过lua执行完成的

protected RFuture<Long> pushTaskAsync() {
    return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_LONG,
            "local expiredValues = redis.call('zrangebyscore', KEYS[2], 0, ARGV[1], 'limit', 0, ARGV[2]); "
          + "if #expiredValues > 0 then "
              + "for i, v in ipairs(expiredValues) do "
                  + "local randomId, value = struct.unpack('dLc0', v);"
                  + "redis.call('rpush', KEYS[1], value);"
                  + "redis.call('lrem', KEYS[3], 1, v);"
              + "end; "
              + "redis.call('zrem', KEYS[2], unpack(expiredValues));"
          + "end; "
            // get startTime from scheduler queue head task
          + "local v = redis.call('zrange', KEYS[2], 0, 0, 'WITHSCORES'); "
          + "if v[1] ~= nil then "
             + "return v[2]; "
          + "end "
          + "return nil;",
          Arrays.<Object>asList(getRawName(), timeoutSetName, queueName),
          System.currentTimeMillis(), 100);
}
复制代码

更加详细的流程大家可以去观看一下源码，这里只是将主要的流程简单介绍了一下。

整合Springboot

以下为maven依赖，将此依赖加入pom文件中

<dependency>
     <groupId>org.redisson</groupId>
     <artifactId>redisson-spring-boot-starter</artifactId>
     <version>3.15.5</version>
</dependency>
复制代码

项目配置中心添加redis配置

spring.redis.host = test.database7401.scsite.net
spring.redis.port = 7401
spring.redis.password = Vbylce3g5N0lbPbd3f4y854C
复制代码

至此Springboot集成redisson完毕。

代码实现

我们期望业务方更多的关注逻辑开发，只需要编写延时任务的具体执行内容，在需要的节点提交该任务即可。

因此业务方在编写具体延时任务时只需要实现我们提供出来的某一个接口，提交的时候调用某一个方法，向该方法传递延时任务执行所需参数、要执行的类、延时时间、时间单位。基于这个思路我们的具体实现如下：

延时任务使用

延时任务接口

/**
 * @author lindj
 * @date 2021/5/26 3:21 下午
 * @describe 延时任务接口
 **/
public interface DelayJob<T> {
    /**
     * 延时任务接口
     * @param deplayJobDTO
     */
    void execute(DeplayJobDTO<T> deplayJobDTO);
}
复制代码

在需要延时任务的场景，业务方只需要实现这个接口即可。

延时任务执行参数

/**
 * @author lindj
 * @date 2021/5/26 3:15 下午
 * @describe 延时任务参数
 **/
@Data
@NoArgsConstructor
public class DeplayJobDTO<T> implements Serializable {
    /**
     * 延时任务执行所需参数
     */
    private T param;
    /**
     * 延时任务执行类
     */
    private Class clazz;
}
复制代码

其中param为具体业务执行所需的参数，clazz为DelayJob的具体实现类。

延时任务提交接口，业务不需要关注具体的实现，只需要在业务节点调用该接口即可。

@Autowired
private DelayJobProducer delayJobProducer;
this.delayJobProducer.submitDelayJob(deplayJobDTO, 30L,
        TimeUnit.SECONDS);
复制代码

至此关于延时队列这部分，业务方只需要关注这么多。下面我们来看一下，延时任务提交类以及提交的延时任务后台是怎么处理的。个人觉得这部分可以封装在组件中或集成到框架中供多个团队使用。

延时任务提交类实现

延时队列定义，这里我们将队列交给spring管理

@Bean
public RDelayedQueue getDelayedQueue(){
    RBlockingQueue blockingQueue = redissonClient.getBlockingQueue(deleyQueueName);
    return redissonClient.getDelayedQueue(blockingQueue);
}
复制代码

延时任务提交接口

/**
 * @author lindj
 * @date 2021/5/26 3:48 下午
 * @describe 延时任务提交接口
 **/
public interface DelayJobProducer {
    /**
     * 提交延时任务
     * @param deplayJobDTO
     * @param delayTime
     * @param timeUnit
     */
    void submitDelayJob(DeplayJobDTO deplayJobDTO, Long delayTime,
                        TimeUnit timeUnit);
}
复制代码

延时任务提交接口实现

/**
 * @author lindj
 * @date 2021/5/26 3:49 下午
 * @describe 延时任务提交类
 **/
@Component
public class DelayJobProducerImpl implements DelayJobProducer {
    @Autowired
    private RDelayedQueue delayedQueue;
    /**
     *
     * @param deplayJobDTO
     * @param delayTime
     * @param timeUnit
     */
    @Override
    public void submitDelayJob(DeplayJobDTO deplayJobDTO, Long delayTime, TimeUnit timeUnit) {
        delayedQueue.offer(deplayJobDTO,delayTime,timeUnit);
    }
}
复制代码

延时任务的执行

大致的思路就是，节点提交完延时任务之后，后台会有一个线程从队列中不断的取出任务进行消费，根据延时任务参数中指定的class，从spring容器中找到该对象，并提交给线程池执行execute方法。具体代码如下：

/**
 * @author lindj
 * @date 2021/5/26 3:33 下午
 * @describe 延时任务执行类
 **/
@Component
@Slf4j
public class DelayJobTimmer {
    @Value("${redisson.delayJob.queueName}")
    private String deleyQueueName;
    @Autowired
    private RedissonClient client;
    @Autowired
    private ApplicationContext context;
    ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
    @PostConstruct
    public void startJobTimer() {
        RBlockingQueue<DeplayJobDTO> blockingQueue =
                client.getBlockingQueue(deleyQueueName);
        new Thread() {
            @Override
            public void run() {
                while (true) {
                    try {
                        // 从队列中获取任务
                        DeplayJobDTO jobDTO = blockingQueue.take();
                        executorService.execute(new DelayJobTask(context, jobDTO));
                    } catch (Exception e) {
                        log.error("延时任务执行失败{}", e);
                        //todo 可以保存数据库
                    }
                }
            }
        }.start();
    }
    
    class DelayJobTask implements Runnable {
        private ApplicationContext context;
        private DeplayJobDTO deplayJobDTO;
        public DelayJobTask(ApplicationContext context, DeplayJobDTO deplayJobDTO) {
            this.context = context;
            this.deplayJobDTO = deplayJobDTO;
        }
        @Override
        public void run() {
            DelayJob delayJob = (DelayJob) context.getBean(deplayJobDTO.getClazz());
            delayJob.execute(deplayJobDTO);
        }
    }
}
复制代码

测试

首先我们要新建一个类实现DelayJob接口，延时任务逻辑部分，这里简单打印了一下参数。

/**
 * @author lindj
 * @date 2021/5/26 3:42 下午
 * @describe
 **/
@Component
@Slf4j
public class TestDelayJob implements DelayJob<ParamDTO> {
    @Override
    public void execute(DeplayJobDTO<ParamDTO> deplayJobDTO) {
        ParamDTO paramDTO = deplayJobDTO.getParam();
       log.info("TestDelayJobService starting paramDTO={}", paramDTO);
    }
}
复制代码

这里定义了一个接口，设置延时任务参数以及具体的实现类class，设定任务30秒后执行。

@GetMapping(value = "/api/delayJob/producer")
public String setValue(){
    DeplayJobDTO<ParamDTO> deplayJobDTO = new DeplayJobDTO<>();
    ParamDTO paramDTO = new ParamDTO();
    paramDTO.setName("lindj");
    paramDTO.setAge(66);
    
    deplayJobDTO.setParam(paramDTO);
    
    deplayJobDTO.setClazz(TestDelayJob.class);
    this.delayJobProducer.submitDelayJob(deplayJobDTO, 30L,
            TimeUnit.SECONDS);
    log.info("延时任务提交完成");
    return "success";
}
复制代码