<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>${lastest.version}</version>
</dependency>

@Service
public class RedissonQueue {

    public static final String QUEUE = "delayQueue";

    // 默認(rèn)超時時間，30秒
    public static final Integer DEFAULT_TIMEOUT = 30;

    @Resource
    private RedissonClient redissonClient;

    // 加入任務(wù)并設(shè)置到期時間
    public void offer(String taskId, Integer timeout) {
        RDelayedQueue<String> delayedQueue = delayedQueue();
        delayedQueue.offer(taskId, Objects.isNull(timeout) ? DEFAULT_TIMEOUT : timeout, TimeUnit.SECONDS);
    }

    // 移除任務(wù)
    public void remove(String taskId) {
        RDelayedQueue<String> delayedQueue = delayedQueue();
        delayedQueue.removeIf(messageId -> messageId.equals(taskId));
    }

    // 任務(wù)列表
    public RDelayedQueue<String> delayedQueue() {
        RBlockingDeque<String> blockingDeque = blockingDeque();
        return redissonClient.getDelayedQueue(blockingDeque);
    }

    public RBlockingDeque<String> blockingDeque() {
        return redissonClient.getBlockingDeque(QUEUE);
    }

    public boolean isShutdown() {
        return redissonClient.isShutdown();
    }

    public void shutdown() {
        redissonClient.shutdown();
    }

}

@Slf4j
@Service
public class RedissonService implements ApplicationRunner {

    @Resource
    private RedissonQueue redissonQueue;

    @Resource(name = "threadPoolTaskExecutor")
    private ThreadPoolTaskExecutor executor;

    @Override
    public void run(ApplicationArguments args) {
        RBlockingDeque<String> blockingDeque = redissonQueue.blockingDeque();
        executor.execute(() -> {
            while (true) {
                if (redissonQueue.isShutdown()) {
                    return;
                } else {
                    String messageId = null;
                    try {
                        messageId = blockingDeque.take();
                    } catch (InterruptedException e) {
                        log.warn("RedissonConsumer error:{}", e.getMessage());
                    }
                    if (!Objects.isNull(messageId) && !messageId.isEmpty()) {
                        log.warn("timeout messageId : {}", messageId);
                    }
                }
            }
        });

    }

    // 初始化，啟動服務(wù)就執(zhí)行一次
    @PostConstruct
    public void init() {
        redissonQueue.delayedQueue();
    }

    @PreDestroy
    public void shutdown() {
        redissonQueue.shutdown();
    }

}

@Operation(summary = "添加任務(wù)", description = "添加任務(wù)")
@PostMapping
public ResponseEntity<?> add(@RequestParam(value = "taskId", required = false) String taskId,
                             @RequestParam(value = "timeout", required = false) Integer timeout) {
    taskId = StringUtils.isEmpty(taskId) ? String.valueOf(snowflake.nextId()) : taskId;
    redissonQueue.offer(taskId, timeout);
    return ResponseEntity.ok().body(redissonQueue.delayedQueue());
}

@Operation(summary = "移除任務(wù)", description = "移除任務(wù)")
@DeleteMapping(value = "/{taskId}")
public ResponseEntity<?> remove(@PathVariable("taskId") String taskId) {
    redissonQueue.remove(taskId);
    return ResponseEntity.ok().body(redissonQueue.delayedQueue());
}