1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

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

    }

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

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;

    }

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

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

    }

}

1

2

3

4

5

private void executeTask(String taskJson) {

    JSONObject task = JSONObject.parseObject(taskJson);

    // 在此处添加具体的业务逻辑

    log.info("执行任务: " + task);

}