Redis实现延时队列.md · 浪子花梦/技术文章收录

## 使用背景
将每个箱子搬运到车子上面，一个箱子相当于一个搬运任务，满n个或者到了一定的时间自动触发车子的搬运任务
## 导入依赖包

```java
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.11.4</version>
</dependency>
```

## 添加Redission配置类

```java
@Slf4j
@Configuration
public class RedissionConfig {
    private final String REDISSON_PREFIX = "redis://";
    private final RedisProperties redisProperties;

public RedissionConfig(RedisProperties redisProperties) {
        this.redisProperties = redisProperties;
    }

@Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        String url = REDISSON_PREFIX + redisProperties.getHost() + ":" + redisProperties.getPort();
        // 这里以单台redis服务器为例
        config.useSingleServer()
                .setAddress(url)
                .setPassword(redisProperties.getPassword())
                .setDatabase(redisProperties.getDatabase())
                .setPingConnectionInterval(2000);
        config.setLockWatchdogTimeout(10000L);

// 实际开发过程中应该为cluster或者哨兵模式，这里以cluster为例
        // String[] urls = {"127.0.0.1:6379", "127.0.0.2:6379"};
        // config.useClusterServers()
        //        .addNodeAddress(urls);
        try {
            return Redisson.create(config);
        } catch (Exception e) {
            log.error("RedissonClient init redis url:[{}], Exception:", url, e);
            return null;
        }
    }
}
```
## 添加延时队列相关类
### RedisDelayQueueUtil 延时队列工具类
```java
@Slf4j
@Component
public class RedisDelayQueueUtil {

@Autowired
    private RedissonClient redissonClient;

/**
     * 添加延迟队列
     *
     * @param value     队列值
     * @param delay     延迟时间
     * @param timeUnit  时间单位
     * @param queueCode 队列键
     * @param <T>
     */
    public <T> boolean addDelayQueue(T value, long delay, TimeUnit timeUnit, String queueCode) {
        if (StringUtils.isBlank(queueCode) || Objects.isNull(value)) {
            return false;
        }
        try {
            redissonClient.getDelayedQueue(redissonClient.getBlockingDeque(queueCode)).poll();
            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("(添加延时队列失败)");
        }
        return true;
    }

/**
     * 获取延迟队列
     *
     * @param queueCode 队列键
     * @param <T>
     */
    public <T> T getDelayQueue(String queueCode) throws InterruptedException {
        if (StringUtils.isBlank(queueCode)) {
            return null;
        }
        RBlockingDeque<Map> blockingDeque = redissonClient.getBlockingDeque(queueCode);
        T value = (T) blockingDeque.poll();
        return value;
    }

/**
     * 删除延时队列
     *
     * @param queueCode 队列键
     */
    public void removeDelayQueue(String queueCode) {
        redissonClient.getDelayedQueue(redissonClient.getBlockingDeque(queueCode)).poll();
    }
}
```

### RedisDelayQueueRunner 启动延迟队列监测扫描

```java
@Slf4j
@Component
public class RedisDelayQueueRunner implements CommandLineRunner {

@Resource
    private RedisDelayQueueUtil redisDelayQueueUtil;
    @Resource
    private ApplicationContext context;
    @Resource
    private ThreadPoolTaskExecutor ptask;

private ThreadPoolExecutor executorService = new ThreadPoolExecutor(3, 5, 30, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1000),new ThreadFactoryBuilder().setNameFormat("task-delay-%d").build());

@Override
    public void run(String... args) throws Exception {
        ptask.execute(() -> {
            while (true){
                try {
                    RedisDelayQueueEnum[] queueEnums = RedisDelayQueueEnum.values();
                    for (RedisDelayQueueEnum queueEnum : queueEnums) {
                        Object value = redisDelayQueueUtil.getDelayQueue(queueEnum.getCode());
                        if (value != null) {
                            RedisDelayQueueHandle<Object> redisDelayQueueHandle = (RedisDelayQueueHandle<Object>)context.getBean(queueEnum.getBeanId());
                            executorService.execute(() -> {redisDelayQueueHandle.execute(value);});
                        }
                    }
                    TimeUnit.MILLISECONDS.sleep(500);
                } catch (InterruptedException e) {
                    log.error("(Redission延迟队列监测异常中断) {}", e.getMessage());
                }
            }
        });
        log.info("(Redission延迟队列监测启动成功)");
    }
}
```

### RedisDelayQueueHandle 启动延迟队列监测扫描

```java
public interface RedisDelayQueueHandle<T> {

void execute(T t);
}
```

### RedisDelayQueueEnum 延迟队列业务枚举

```java
@Getter
@NoArgsConstructor
@AllArgsConstructor
public enum RedisDelayQueueEnum {

/**
     * 输送线搬运任务池，系统自动下发任务
     */
    KMRES_LINE_NOT_FULL_TIMEOUT("KMRES_LINE_NOT_FULL_TIMEOUT", "kmresLineNotFullTimeout");

/**
     * 延迟队列 Redis Key
     */
    private String code;

/**
     * 延迟队列具体业务实现的 Bean
     * 可通过 Spring 的上下文获取
     */
    private String beanId;
}
```
### KmresLineNotFullTimeout任务池未满，系统自动下发任务

```java
@Slf4j
@Component
public class KmresLineNotFullTimeout implements RedisDelayQueueHandle<String> {

@Autowired
    private PoolTaskHandler poolTaskHandler;

/**
     * 执行KMRES任务下发
     *
     * @param redisQueueName REDIS任务池名称
     */
    @Override
    public void execute(String redisQueueName) {
        poolTaskHandler.handleKmresTask(redisQueueName);
        log.info("<<<<<<<<<<<<<<<<<<<<<<< RcsNotFullTimeout 延时任务执行成功.");
    }
}
```

### PoolTaskHandler 任务池：当监听到消息时，等待下发任务

```java
@Component
public class PoolTaskHandler {

@Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private RedisDelayQueueUtil redisDelayQueueUtil;

/**
     * 执行搬运任务
     *
     * @param taskList        需要下发的任务集合
     * @param redisQueueName  任务池类型         <{@link CommonConstants.REDIS_TASK_POOL_TYPE}/>
     * @param taskPoolMaxSize 任务池各类型最大值 <{@link CommonConstants.REDIS_TASK_POOL_MAX_SIZE}/>
     * @param delayQueueName  延时队列类型       <{@link CommonConstants.REDIS_DELAY_QUEUE_TYPE}/>
     */
    public void execute(List<BusinessTask> taskList, String redisQueueName, long taskPoolMaxSize, String delayQueueName) {
        if (CollectionUtils.isEmpty(taskList)) {
            return;
        }
        for (BusinessTask t : taskList) {
            // 将任务存放至redis队列中
            redisTemplate.boundListOps(redisQueueName).leftPush(JSONUtils.toJSONString(t));

// 判断是否达到下发任务的条件：满足5个元素
            long queueSize = redisTemplate.boundListOps(redisQueueName).size();
            if (queueSize >= taskPoolMaxSize) {
                // 处理Redis数据，调用RSC接口下发任务
                handleKmresTask(redisQueueName);
                // 删除延时任务
                redisDelayQueueUtil.removeDelayQueue(delayQueueName);
            } else {
                // 更新延时任务
                redisDelayQueueUtil.addDelayQueue(redisQueueName, CommonConstants.DELAY_TIMEOUT, TimeUnit.SECONDS, delayQueueName);
            }
        }
    }

/**
     * 执行KMRES任务下发
     *
     * @param redisQueueName 任务池类型
     */
    public void handleKmresTask(String redisQueueName) {

// TODO: 下发KMRES接口逻辑编写 ...

}
}
```

### 使用方式
调用 PoolTaskHandler 里的 execute 方法

浪子花梦/技术文章收录

简介

发行版

贡献者

近期动态

浪子花梦/技术文章收录 .gitee-modal { width: 500px !important; }

简介

发行版

贡献者

近期动态

搜索帮助

浪子花梦/技术文章收录