【原创】分布式之延时任务方案解析

CREATE TABLE `delay_task` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `biz_id` varchar(64) NOT NULL COMMENT '业务ID，如订单ID',
  `biz_type` varchar(32) NOT NULL COMMENT '业务类型，如ORDER_TIMEOUT',
  `execute_time` datetime NOT NULL COMMENT '计划执行时间',
  `status` tinyint(4) NOT NULL DEFAULT '0' COMMENT '状态：0-待执行，1-执行中，2-已完成，3-已取消',
  `params` text COMMENT '任务执行所需参数',
  `created_time` datetime DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `idx_execute_time_status` (`execute_time`,`status`) -- 核心索引
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

@Component
public class DelayQueueManager {
    private final DelayQueue<DelayTask> delayQueue = new DelayQueue<>();
 
    // 添加任务
    public void addTask(DelayTask task) {
        delayQueue.put(task);
    }
 
    @PostConstruct
    public void startConsumer() {
        new Thread(() -> {
            while (true) {
                try {
                    DelayTask task = delayQueue.take();
                    // 执行任务逻辑
                    task.execute();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }).start();
    }
}
 
// 任务类
@Data
public class DelayTask implements Delayed {
    private String id;
    private String bizData;
    private long executeTime; // 执行时间戳（毫秒）
 
    public DelayTask(String id, String bizData, long delayMs) {
        this.id = id;
        this.bizData = bizData;
        this.executeTime = System.currentTimeMillis() + delayMs;
    }
 
    // 获取剩余延迟时间
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(executeTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }
 
    // 用于队列内部排序
    @Override
    public int compareTo(Delayed o) {
        return Long.compare(this.executeTime, ((DelayTask) o).executeTime);
    }
 
    public void execute() {
        // 具体的业务逻辑
        System.out.println("执行任务: " + bizData);
    }
}

@Component
public class RedisDelayService {
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
 
    private static final String KEY = "DELAY_TASKS";
 
    // 添加延时任务
    public void addTask(String taskId, long delayMs) {
        long executeTime = System.currentTimeMillis() + delayMs;
        redisTemplate.opsForZSet().add(KEY, taskId, executeTime);
    }
 
    @PostConstruct
    public void startPolling() {
        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
        executor.scheduleAtFixedRate(() -> {
            long now = System.currentTimeMillis();
            // 1. 获取所有到期的任务
            Set<String> taskIds = redisTemplate.opsForZSet().rangeByScore(KEY, 0, now);
            if (taskIds != null && !taskIds.isEmpty()) {
                for (String taskId : taskIds) {
                    // 2. 尝试移除该任务，移除成功的那个实例获得执行权
                    Long removed = redisTemplate.opsForZSet().remove(KEY, taskId);
                    if (removed != null && removed > 0) {
                        // 3. 执行任务
                        processTask(taskId);
                    }
                }
            }
        }, 0, 1, TimeUnit.SECONDS); // 每秒轮询一次
    }
 
    private void processTask(String taskId) {
        // 具体的任务处理逻辑
        System.out.println("Processing task: " + taskId);
    }
}

@Service
public class OrderService {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
 
    public void createOrder(Order order) {
        // ... 保存订单等逻辑
 
        // 发送一条延迟消息，30分钟后检查订单支付状态
        Message<String> message = MessageBuilder.withPayload(order.getId())
                .setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, "4") // 级别4对应30分钟
                .build();
        rocketMQTemplate.send("ORDER_TIMEOUT_TOPIC", message);
    }
 
    // 消费者
    @RocketMQMessageListener(topic = "ORDER_TIMEOUT_TOPIC", consumerGroup = "order-timeout-group")
    public class OrderTimeoutConsumer implements RocketMQListener<String> {
        @Override
        public void onMessage(String orderId) {
            // 检查订单是否支付，未支付则执行取消逻辑
            checkAndCancelOrder(orderId);
        }
    }
}