当前位置：首页 > news >正文

redission实现延时队列

news 2025/11/4 9:33:14

自用，分布式的延时队列

效果

前提代码

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

spring:data:redis:host: localhostpassword: 123456

@Getter
@ToString
public class DelayedTask<T> implements Delayed, Serializable {private static final long serialVersionUID = 3018458065076640934L;private final T taskContent;private final Long triggerTime;/*** @param seconds 秒*/public DelayedTask(T taskContent, Long seconds) {this.taskContent = taskContent;this.triggerTime = System.currentTimeMillis() + seconds * 1000;}@Overridepublic long getDelay(TimeUnit unit) {return unit.convert(triggerTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);}@Overridepublic int compareTo(Delayed o) {return this.triggerTime.compareTo(((DelayedTask) o).triggerTime);}}

核心代码

@RestController
@RequiredArgsConstructor
@RequestMapping("/")
@Slf4j
public class DelayController {private final RedissonClient redissonClient;private volatile boolean QUEUE_RUNNING = true;@GetMapping("/test")public void test() throws InterruptedException {RBlockingDeque<DelayedTask> blockingDeque = redissonClient.getBlockingDeque("my-delay-queue");RDelayedQueue<DelayedTask> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);new Thread(()->{while (QUEUE_RUNNING){try {var delayedTask = blockingDeque.take();log.info("需要执行这个任务,{}", delayedTask);} catch (Exception e) {log.error("延时任务出错：{}", e);}}}).start();Thread.sleep(15000L);QUEUE_RUNNING = false;}@PostMapping("/add")public void add() {RBlockingDeque<DelayedTask> blockingDeque = redissonClient.getBlockingDeque("my-delay-queue");final RDelayedQueue<DelayedTask> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);delayedQueue.offer(new DelayedTask("做这个任务", 3L),3L, TimeUnit.SECONDS);delayedQueue.offer(new DelayedTask("做这个任务2",5L),5L, TimeUnit.SECONDS);delayedQueue.offer(new DelayedTask("做这个任务3",8L),8L, TimeUnit.SECONDS);}
}