Redis ZSet 实现延时告警重推：从需求出发的架构选型思考

import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

public class AlertTask implements Delayed {

    private final String alertId;
    private final long triggerTime; // 触发时间戳（毫秒）

    public AlertTask(String alertId, long delayMs) {
        this.alertId = alertId;
        this.triggerTime = System.currentTimeMillis() + delayMs;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        long diff = triggerTime - System.currentTimeMillis();
        return unit.convert(diff, TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed other) {
        return Long.compare(
            this.triggerTime,
            ((AlertTask) other).triggerTime
        );
    }

    public String getAlertId() {
        return alertId;
    }
}

import java.util.concurrent.DelayQueue;

@Component
public class InMemoryAlertScheduler {

    private final DelayQueue<AlertTask> queue = new DelayQueue<>();

    @PostConstruct
    public void startConsumer() {
        Thread consumer = new Thread(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    AlertTask task = queue.take(); // 阻塞直到有到期任务
                    handleAlert(task.getAlertId());
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }, "alert-delay-consumer");
        consumer.setDaemon(true);
        consumer.start();
    }

    public void schedule(String alertId, long delayMs) {
        queue.put(new AlertTask(alertId, delayMs));
    }

    private void handleAlert(String alertId) {
        // 查询告警是否已确认，未确认则重推
    }
}

@Service
public class AlertQueueService {

    private static final String QUEUE_KEY = "alert:delay:queue";

    @Autowired
    private StringRedisTemplate redisTemplate;

    /**
     * 将告警加入延时队列
     *
     * @param alertId  告警 ID
     * @param delayMs  延迟毫秒数
     */
    public void enqueue(String alertId, long delayMs) {
        double triggerTime = System.currentTimeMillis() + delayMs;
        redisTemplate.opsForZSet().add(QUEUE_KEY, alertId, triggerTime);
    }

    /**
     * 告警被确认时，从队列中移除
     */
    public void cancel(String alertId) {
        redisTemplate.opsForZSet().remove(QUEUE_KEY, alertId);
    }

    /**
     * 查询待处理任务数
     */
    public Long pendingCount() {
        return redisTemplate.opsForZSet().zCard(QUEUE_KEY);
    }
}

@Component
public class AlertScheduler {

    private static final String QUEUE_KEY = "alert:delay:queue";
    private static final String LOCK_KEY = "alert:scheduler:lock";

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private AlertNotifyService notifyService;

    /**
     * Lua 脚本：原子性地取出到期任务并删除
     * 避免多节点重复消费
     */
    private static final String FETCH_AND_REMOVE_LUA =
        "local tasks = redis.call('ZRANGEBYSCORE', KEYS[1], '0', ARGV[1], 'LIMIT', 0, ARGV[2])\n" +
        "if #tasks > 0 then\n" +
        "    redis.call('ZREM', KEYS[1], unpack(tasks))\n" +
        "end\n" +
        "return tasks";

    private final DefaultRedisScript<List> fetchScript;

    public AlertScheduler() {
        this.fetchScript = new DefaultRedisScript<>();
        this.fetchScript.setScriptText(FETCH_AND_REMOVE_LUA);
        this.fetchScript.setResultType(List.class);
    }

    /**
     * 每 10 秒轮询一次，取出到期告警任务
     */
    @Scheduled(fixedDelay = 10_000)
    public void pollExpiredAlerts() {
        // 分布式锁：防止多节点同时消费
        Boolean locked = redisTemplate.opsForValue()
            .setIfAbsent(LOCK_KEY, "1", Duration.ofSeconds(30));
        if (!Boolean.TRUE.equals(locked)) {
            return; // 其他节点正在处理
        }

        try {
            String now = String.valueOf(System.currentTimeMillis());
            @SuppressWarnings("unchecked")
            List<String> alertIds = redisTemplate.execute(
                fetchScript,
                List.of(QUEUE_KEY),
                now, "50" // 每次最多取 50 条
            );

            if (alertIds != null && !alertIds.isEmpty()) {
                for (String alertId : alertIds) {
                    processAlert(alertId);
                }
            }
        } finally {
            redisTemplate.delete(LOCK_KEY);
        }
    }

    private void processAlert(String alertId) {
        try {
            // 查询告警是否已确认
            if (!notifyService.isAcknowledged(alertId)) {
                notifyService.resend(alertId);
            }
        } catch (Exception e) {
            // 处理失败，重新入队（延迟 60 秒重试）
            redisTemplate.opsForZSet().add(
                QUEUE_KEY,
                alertId,
                System.currentTimeMillis() + 60_000
            );
        }
    }
}

@SpringBootApplication
@EnableScheduling
public class AlertApplication {
    public static void main(String[] args) {
        SpringApplication.run(AlertApplication.class, args);
    }
}

@Configuration
public class RabbitAlertConfig {

    @Bean
    public Queue delayQueue() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange", "alert.dlx");
        args.put("x-dead-letter-routing-key", "alert.process");
        return new Queue("alert.delay.queue", true, false, false, args);
    }

    @Bean
    public DirectExchange deadLetterExchange() {
        return new DirectExchange("alert.dlx");
    }

    @Bean
    public Queue processQueue() {
        return new Queue("alert.process.queue");
    }

    @Bean
    public Binding dlxBinding() {
        return BindingBuilder
            .bind(processQueue())
            .to(deadLetterExchange())
            .with("alert.process");
    }
}

@Service
public class RabbitAlertProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void scheduleAlert(String alertId, long delayMs) {
        rabbitTemplate.convertAndSend(
            "", "alert.delay.queue", alertId,
            message -> {
                message.getMessageProperties()
                    .setExpiration(String.valueOf(delayMs));
                return message;
            }
        );
    }
}

// 按天分 Key，避免单个 ZSet 无限膨胀
String queueKey = "alert:delay:queue:" + LocalDate.now().format(
    DateTimeFormatter.ofPattern("yyyyMMdd")
);

// 定时清理过期 Key（凌晨清理前一天的 Key）
@Scheduled(cron = "0 0 3 * * ?")
public void cleanExpiredKeys() {
    String yesterday = LocalDate.now().minusDays(1).format(
        DateTimeFormatter.ofPattern("yyyyMMdd")
    );
    redisTemplate.delete("alert:delay:queue:" + yesterday);
}

@Scheduled(fixedDelay = 10_000)
public void pollExpiredAlerts() {
    try {
        // 正常 ZSet 消费逻辑
        doPoll();
    } catch (RedisConnectionFailureException e) {
        log.error("Redis 连接失败，启用降级策略");
        // 降级：从 MySQL 查询未确认且超时的告警
        fallbackFromMySQL();
    }
}

private void fallbackFromMySQL() {
    List<Alert> expired = alertMapper.selectUnacknowledgedBefore(
        LocalDateTime.now().minusMinutes(5)
    );
    for (Alert alert : expired) {
        notifyService.resend(alert.getId());
    }
}

RBlockingQueue<String> blockingQueue = redisson.getBlockingQueue("alert-queue");
RDelayedQueue<String> delayedQueue = redisson.getDelayedQueue(blockingQueue);

// 入队
delayedQueue.offer("alert-123", 5, TimeUnit.MINUTES);

// 消费
String alertId = blockingQueue.take(); // 阻塞式

private void processAlert(String alertId) {
    // 1. 先查询数据库中的告警状态
    Alert alert = alertMapper.selectById(alertId);
    if (alert == null || alert.isAcknowledged()) {
        return; // 已确认，跳过
    }
    // 2. CAS 更新状态，防止并发重复处理
    int updated = alertMapper.updateStatusCAS(
        alertId, AlertStatus.PENDING, AlertStatus.RESENDING
    );
    if (updated == 0) {
        return; // 已被其他线程处理
    }
    // 3. 执行重推
    notifyService.resend(alertId);
}