Redis模拟延时队列实现日程提醒的方法

private static final String ZSET_KEY = "sms_delayed_tasks";
private JedisCluster jedisCluster;
public void RedisClusterScheduler() {
    Set<HostAndPort> nodes = new HashSet<>();
    //从配置文件中读取redis集群配置
    for (String node : AcpCore.getProp("spring.redis.cluster.nodes").split(",")) {
        String[] hostPort = node.split(":");
        nodes.add(new HostAndPort(hostPort[0], Integer.parseInt(hostPort[1])));
    }
    GenericObjectPoolConfig<Jedis> poolConfig = new GenericObjectPoolConfig<>();
    poolConfig.setMaxTotal(128);
    poolConfig.setMaxIdle(128);
    poolConfig.setMinIdle(16);
    jedisCluster = new JedisCluster(nodes, 2000, 2000, 5, AcpCore.getProp("spring.redis.password"), poolConfig);
    if (!isCalled) {
        isCalled = true;
        startTaskChecker();
    }
}

public void addDelayedTask(String task, String time) {
    long executeTime = convertToTimestamp(time);
    if (executeTime > System.currentTimeMillis() / 1000) {
        jedisCluster.zadd(ZSET_KEY, executeTime, task);
        log.info("添加任务: " + task + " 执行时间: " + executeTime);
    } else {
        log.info("任务时间必须在当前时间之后: " + task);
    }
}
private long convertToTimestamp(String time) {
    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    try {
        return sdf.parse(time).getTime() / 1000;
    } catch (ParseException e) {
        e.printStackTrace();
        return System.currentTimeMillis() / 1000;
    }
}

private void startTaskChecker() {
    executorService.submit(() -> {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                checkAndExecuteTasks();
                Thread.sleep(1000);
            } catch (Exception e) {
                log.info(new Date() + "发生异常但不中断，异常是：" + e);
            }
        }
    });
}
private void checkAndExecuteTasks() {
    long currentTime = System.currentTimeMillis() / 1000;
    Set<String> tasks = jedisCluster.zrangeByScore(ZSET_KEY, 0, currentTime);
    for (String task : tasks) {
        jedisCluster.zrem(ZSET_KEY, task);
        executeTask(task);
    }
}

private void executeTask(String taskJson) {
    JSONObject task = JSONObject.parseObject(taskJson);
    // 在此处添加具体的业务逻辑
    log.info("执行任务: " + task);
}