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);
    }
}

@Service
public class AlertMergeService {

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

    private static final String ENQUEUE_LUA = """
        local version = redis.call('HINCRBY', KEYS[2], ARGV[1], 1)
        redis.call('ZADD', KEYS[1], ARGV[2], ARGV[1])
        return version
        """;

    private final StringRedisTemplate redisTemplate;
    private final AlertDetailMapper alertDetailMapper;
    private final AlertMergeMapper alertMergeMapper;

    public AlertMergeService(
        StringRedisTemplate redisTemplate,
        AlertDetailMapper alertDetailMapper,
        AlertMergeMapper alertMergeMapper
    ) {
        this.redisTemplate = redisTemplate;
        this.alertDetailMapper = alertDetailMapper;
        this.alertMergeMapper = alertMergeMapper;
    }

    public long enqueue(AlertEvent event) {
        String mergeKey = buildMergeKey(event);
        long triggerTime = System.currentTimeMillis() + event.delayMs();

        // 先落库，保证后续归并可以按 mergeKey 找回完整明细。
        alertDetailMapper.insert(event.toEntity(mergeKey));

        Long version = redisTemplate.execute(
            RedisScript.of(ENQUEUE_LUA, Long.class),
            List.of(QUEUE_KEY, VERSION_KEY),
            mergeKey,
            String.valueOf(triggerTime)
        );
        return version == null ? 0L : version;
    }
}

public record MergeTaskSnapshot(
    String mergeKey,
    double score,
    long version
) {
}

public List<MergeTaskSnapshot> fetchExpired(long now, int limit) {
    Set<ZSetOperations.TypedTuple<String>> tuples = redisTemplate
        .opsForZSet()
        .rangeByScoreWithScores(QUEUE_KEY, 0, now, 0, limit);

    if (tuples == null || tuples.isEmpty()) {
        return List.of();
    }

    List<MergeTaskSnapshot> snapshots = new ArrayList<>();
    for (ZSetOperations.TypedTuple<String> tuple : tuples) {
        String mergeKey = tuple.getValue();
        Double score = tuple.getScore();
        Object version = redisTemplate.opsForHash()
            .get(VERSION_KEY, mergeKey);

        if (mergeKey != null && score != null && version != null) {
            snapshots.add(new MergeTaskSnapshot(
                mergeKey,
                score,
                Long.parseLong(version.toString())
            ));
        }
    }
    return snapshots;
}

public void mergeBySnapshot(MergeTaskSnapshot snapshot) {
    List<AlertDetail> details = alertDetailMapper
        .selectPendingByMergeKey(snapshot.mergeKey());
    if (details.isEmpty()) {
        return;
    }

    AlertMergeResult result = AlertMergeResult.from(
        snapshot.mergeKey(),
        details
    );

    int updated = alertMergeMapper.upsertByMergeKey(result);
    if (updated == 0) {
        return;
    }

    alertDetailMapper.markMerged(snapshot.mergeKey());
}

private static final String SAFE_ACK_LUA = """
    local currentScore = redis.call('ZSCORE', KEYS[1], ARGV[1])
    if not currentScore then
        return 0
    end

    local currentVersion = redis.call('HGET', KEYS[2], ARGV[1])
    if tonumber(currentScore) == tonumber(ARGV[2])
        and currentVersion == ARGV[3] then
        return redis.call('ZREM', KEYS[1], ARGV[1])
    end

    return 0
    """;

public boolean safeAck(MergeTaskSnapshot snapshot) {
    Long removed = redisTemplate.execute(
        RedisScript.of(SAFE_ACK_LUA, Long.class),
        List.of(QUEUE_KEY, VERSION_KEY),
        snapshot.mergeKey(),
        String.valueOf(snapshot.score()),
        String.valueOf(snapshot.version())
    );
    return Long.valueOf(1L).equals(removed);
}

private static final String SAFE_RETRY_LUA = """
    local currentScore = redis.call('ZSCORE', KEYS[1], ARGV[1])
    if not currentScore then
        return 0
    end

    local currentVersion = redis.call('HGET', KEYS[2], ARGV[1])
    if tonumber(currentScore) == tonumber(ARGV[2])
        and currentVersion == ARGV[3] then
        redis.call('ZADD', KEYS[1], ARGV[4], ARGV[1])
        return 1
    end

    return 0
    """;

public boolean safeRetry(MergeTaskSnapshot snapshot, long nextRetryTime) {
    Long updated = redisTemplate.execute(
        RedisScript.of(SAFE_RETRY_LUA, Long.class),
        List.of(QUEUE_KEY, VERSION_KEY),
        snapshot.mergeKey(),
        String.valueOf(snapshot.score()),
        String.valueOf(snapshot.version()),
        String.valueOf(nextRetryTime)
    );
    return Long.valueOf(1L).equals(updated);
}

@Scheduled(fixedDelay = 10_000)
public void pollExpiredMergeTasks() {
    long now = System.currentTimeMillis();
    for (MergeTaskSnapshot snapshot : fetchExpired(now, 50)) {
        try {
            mergeBySnapshot(snapshot);
            safeAck(snapshot);
        } catch (Exception ex) {
            long nextRetryTime = System.currentTimeMillis() + 60_000;
            safeRetry(snapshot, nextRetryTime);
        }
    }
}

private String buildMergeKey(AlertEvent event) {
    return event.deviceId() + ":" + event.ruleId() + ":" + event.level();
}

@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);
}