破局延时任务(上)：为什么选择Spring Boot + DelayQueue来自研分布式延时队列组件？

匿名上传

发布时间:2025-11-03 14:00:03

1. 背景与需求分析

在现代分布式系统中，延时任务处理是一个常见且关键的需求场景。典型的应用包括：订单超时自动取消、自动确认收货、消息延时推送等。

有趣的是，业务系统中的许多延时任务场景并不严格要求准时执行。以博客平台定时发布文章为例：计划8点40分发布的文章，延迟到8点45分甚至9点发布，通常不会对用户体验造成重大影响，只要最终能够成功发布即可。

然而，从运营角度考虑，如果读者习惯按时阅读你的文章，而文章未能准时发布，可能导致读者误以为今日无更新而错过内容，这种体验是不可接受的。因此，延时任务是否需要准时执行，需根据具体业务场景进行权衡。

下面我们将系统分析分布式环境下延时任务的常见实现方案及其适用场景。

2. 常见实现方案对比

2.1 基于定时任务扫描

对于不要求准时执行的延时任务场景，最直接的实现方式是将任务存储于数据库或Redis的ZSet中（ZSet的score存储执行时间戳），然后通过定时任务周期性扫描。

核心实现逻辑：

// 每分钟扫描一次延时任务
@Scheduled(cron = "0 * * * * *")
public void scheduleScan() {
    // 读取已到期的延时任务
    Set<String> taskIds = stringRedisTemplate.opsForZSet()
        .rangeByScore("task-key", 0, System.currentTimeMillis());
    
    if (CollUtil.isEmpty(taskIds)) {
        return;
    }
    
    taskIds.forEach(taskId -> {
        // 处理延时任务
        ...
    });
    
    // 处理完成后删除任务
    ...
}

方案优缺点：

优点：实现简单，技术门槛低
缺点：
- 无法保证准时执行，延迟时间取决于扫描频率;
- 单机执行模式无法充分利用分布式集群资源
- 任务堆积时处理性能受限

2.2 基于消息队列延时功能

主流消息队列（如RocketMQ、RabbitMQ等）内置了延时队列功能，如果业务系统已使用这类MQ，可以便捷地实现延时任务处理。

但对于使用Kafka等不支持原生延时功能的消息队列的系统，需要自行实现延时逻辑。常见的实现方案是按延时间隔创建多级Topic：

topic_10s：延时10秒执行的队列
topic_1m：延时1分钟执行的队列
topic_30m：延时30分钟执行的队列

实现原理：

生产者根据延时时间将消息发送到对应Topic
消费者轮询消费消息，判断消息产生时间与当前时间的差值
达到预设延时时间则处理消息，否则暂停消费等待

核心代码示例：

kafka消费者配置：

@Configuration
public class KafkaDelayConfig {
    public static final long DELAY_30M = 30 * 60 * 1000L;
    
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "delay-30m-group");
        // 重点，手动提交offset，没到执行时间的不提交，阻塞住
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        return new DefaultKafkaConsumerFactory<>(props);
    }
}

消费者处理延时任务：

@Service
@Slf4j
public class Delay30mConsumer {
    
    @KafkaListener(topics = "topic_30m")
    public void consume30mDelay(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            String messageValue = record.value();
            DelayMessage message = parseMessage(messageValue);
            
            if (message == null) {
                ack.acknowledge();
                return;
            }
            
            long currentTime = System.currentTimeMillis();
            long elapsedTime = currentTime - message.getSendTime();
            
            if (elapsedTime >= KafkaDelayConfig.DELAY_30M) {
                // 达到延时时间，处理任务
                processExpiredMessage(message);
                ack.acknowledge();
                log.info("30分钟延时消息处理完成: {}", message.getId());
            } else {
                // 未到期，暂停消费并等待剩余时间
                TimeUnit.MICROSECONDS.sleep(KafkaDelayConfig.DELAY_30M - elapsedTime);
            }
        } catch (Exception e) {
            log.error("处理30分钟延时消息异常", e);
        }
    }
}

方案适用场景：

延时时间固定的场景（如订单超时未支付自动取消释放库存处理场景）
任务执行顺序有序：先提交的延时任务先到期

局限性：

无法处理动态延时时间的任务（如先提交的延时任务后到期）
Topic数量随延时粒度增加而增多，管理复杂度高

2.3 基于Redis过期Key监听

Redis的键过期机制可以用于实现延时任务：将任务ID作为Key，设置过期时间为延时时间，通过监听过期事件触发任务执行。

实现原理：

@Component
public class MyRedisKeyExpiredEventListener implements ApplicationListener<RedisKeyExpiredEvent> {
    @Override
    public void onApplicationEvent(RedisKeyExpiredEvent event) {
        byte[] body = event.getSource();
        System.out.println("获取到延迟消息：" + new String(body));
        // 执行延时任务处理逻辑
    }
}