延时队列是一种特殊的消息队列，消息在发送后不会立即被消费，而是等待指定的时间后才被消费者处理。就像设置了一个”闹钟”，到时间才响。

阻塞队列 RBlockingDeque - 阻塞双端队列#

特点：

双端：可以从两端插入和取出元素
阻塞：当队列为空的时候，取元素会阻塞等待
线程安全：多个线程可以安全操作

1
// 特点：
2
// - 双端：可以从两端插入和取出元素
3
// - 阻塞：当队列为空时，取元素会阻塞等待
4
// - 线程安全：多个线程可以安全操作
5

6
RBlockingDeque<String> deque = redissonClient.getBlockingDeque("myDeque");
7
deque.offerFirst("头部元素");
8
deque.offerLast("尾部元素");
9
String element = deque.takeFirst(); // 阻塞获取

RDelayedQueue - 延时队列#

特点：

自动延时：消息在指定时间后自动变为可消费状态
精确控制：可以精确控制每个消息的延时时间
Redis实现：基于Redis的有序集合(ZSet)实现

1
RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(deque);
2
delayedQueue.offer("消息内容", 30, TimeUnit.SECONDS); // 30秒后可消费

完整实现示例#

生产者端（消息发送）#

1
@Service
2
public class DelayQueueProducer {
3

4
    @Autowired
5
    private RedissonClient redissonClient;
6

7
    public void sendDelayedMessage(String message, long delaySeconds) {
8
        try {
9
            // 创建队列
10
            RBlockingDeque<String> blockingDeque = redissonClient
11
                .getBlockingDeque("DELAY_QUEUE_EXAMPLE");
12
            RDelayedQueue<String> delayedQueue = redissonClient
13
                .getDelayedQueue(blockingDeque);
14

15
            // 发送延时消息
16
            delayedQueue.offer(message, delaySeconds, TimeUnit.SECONDS);
17

18
            System.out.println("发送延时消息: " + message +
19
                             ", 延时: " + delaySeconds + "秒");
20
        } catch (Exception e) {
21
            log.error("发送延时消息失败", e);
22
        }
23
    }
24
}

消费者端（消息处理）#

1
@Component
2
public class DelayQueueConsumer {
3

4
    @Autowired
5
    private RedissonClient redissonClient;
6

7
    @PostConstruct
8
    public void startConsumer() {
9
        // 启动独立线程消费延时消息
10
        new Thread(this::consumeMessages, "DelayQueueConsumer").start();
11
    }
12

13
    private void consumeMessages() {
14
        try {
15
            RBlockingDeque<String> blockingDeque = redissonClient
16
                .getBlockingDeque("DELAY_QUEUE_EXAMPLE");
17

18
            while (!Thread.currentThread().isInterrupted()) {
19
                // 阻塞获取消息（自动等待延时到期）
20
                String message = blockingDeque.take();
21
                System.out.println("消费延时消息: " + message);
22

23
                // 处理业务逻辑
24
                processMessage(message);
25
            }
26
        } catch (InterruptedException e) {
27
            Thread.currentThread().interrupt();
28
            log.info("消费者线程被中断");
29
        } catch (Exception e) {
30
            log.error("消费消息异常", e);
31
        }
32
    }
33

34
    private void processMessage(String message) {
35
        // 实际的业务处理逻辑
36
        System.out.println("处理业务消息: " + message);
37
    }
38
}

底层实现原理#

Redis数据结构使用#

1
# Redisson使用以下数据结构：
2
# 1. 有序集合(ZSet) - 存储延时消息和到期时间
3
ZADD delay_queue 1640995200 "message1"  # 到期时间戳作为score
4

5
# 2. 列表(List) - 存储已到期可消费的消息
6
LPUSH ready_queue "message1"
7

8
# 3. 定时任务 - 定期检查到期消息
9
# Redisson内部使用定时任务扫描ZSet，将到期消息移动到List

延时检查机制#

1
// Redisson内部逻辑（简化版）
2
public class DelayedQueueChecker {
3
    public void checkExpiredMessages() {
4
        long now = System.currentTimeMillis();
5

6
        // 从有序集合中获取已到期的消息
7
        Set<String> expiredMessages = redisTemplate
8
            .opsForZSet()
9
            .rangeByScore("delay_queue", 0, now);
10

11
        for (String message : expiredMessages) {
12
            // 移动到可消费队列
13
            redisTemplate.opsForList().leftPush("ready_queue", message);
14
            redisTemplate.opsForZSet().remove("delay_queue", message);
15
        }
16
    }
17
}

使用场景#

1
// 1. 订单超时处理
2
public void handleOrderTimeout(String orderId) {
3
    delayedQueue.offer(orderId, 30, TimeUnit.MINUTES);
4
}
5

6
// 2. 优惠券到期提醒
7
public void couponExpireReminder(String couponId) {
8
    delayedQueue.offer(couponId, 24, TimeUnit.HOURS);
9
}
10

11
// 3. 消息重试机制
12
public void messageRetry(String messageId) {
13
    delayedQueue.offer(messageId, 5, TimeUnit.SECONDS); // 5秒后重试
14
}