ZADD delay_tasks <timestamp> <task_id>

import redis.clients.jedis.Jedis;  
import redis.clients.jedis.JedisPubSub;  
  
public class RedisKeyspaceNotifier {  
  
    public static void main(String[] args) {  
        // 創(chuàng)建Jedis實(shí)例  
        Jedis jedis = new Jedis("localhost", 6379);  
  
        // 配置鍵空間通知（通常這一步在Redis配置文件中完成，但也可以在運(yùn)行時(shí)配置）  
        jedis.configSet("notify-keyspace-events", "KEA");  
  
        // 訂閱鍵空間通知  
        jedis.subscribe(new KeyspaceNotificationSubscriber(), "__keyevent@0__:expired");  
    }  
  
    static class KeyspaceNotificationSubscriber extends JedisPubSub {  
  
        @Override  
        public void onMessage(String channel, String message) {  
            System.out.println("Received message from channel: " + channel + ", message: " + message);  
            // 在這里處理接收到的鍵空間通知  
            // 例如，如果message是一個(gè)需要處理的任務(wù)ID，你可以在這里觸發(fā)相應(yīng)的任務(wù)處理邏輯  
        }  
  
        @Override  
        public void onSubscribe(String channel, int subscribedChannels) {  
            System.out.println("Subscribed to channel: " + channel);  
        }  
  
        @Override  
        public void onUnsubscribe(String channel, int subscribedChannels) {  
            System.out.println("Unsubscribed from channel: " + channel);  
        }  
    }  
}

import redis.clients.jedis.Jedis;  
  
import java.util.Set;  
  
public class RedisDelayedTaskDemo {  
  
    private static final String ZSET_KEY = "delayed_tasks";  
    private static final String REDIS_HOST = "localhost";  
    private static final int REDIS_PORT = 6379;  
  
    public static void main(String[] args) {  
        Jedis jedis = new Jedis(REDIS_HOST, REDIS_PORT);  
  
        // 添加延遲任務(wù)  
        addDelayedTask(jedis, "task1", System.currentTimeMillis() / 1000 + 5); // 5秒后執(zhí)行  
        addDelayedTask(jedis, "task2", System.currentTimeMillis() / 1000 + 10); // 10秒后執(zhí)行  
  
        // 模擬定時(shí)任務(wù)檢查器  
        new Thread(() -> {  
            while (true) {  
                try {  
                    // 檢查并執(zhí)行到期的任務(wù)  
                    checkAndExecuteTasks(jedis);  
                    Thread.sleep(1000); // 每秒檢查一次  
                } catch (InterruptedException e) {  
                    e.printStackTrace();  
                }  
            }  
        }).start();  
    }  
  
    private static void addDelayedTask(Jedis jedis, String task, long executeTime) {  
        jedis.zadd(ZSET_KEY, executeTime, task);  
        System.out.println("Added task: " + task + " with execution time: " + executeTime);  
    }  
  
    private static void checkAndExecuteTasks(Jedis jedis) {  
        long currentTime = System.currentTimeMillis() / 1000;  
        Set<String> tasks = jedis.zrangeByScore(ZSET_KEY, 0, currentTime);  
  
        for (String task : tasks) {  
            jedis.zrem(ZSET_KEY, task); // 從有序集合中移除任務(wù)  
            executeTask(task); // 執(zhí)行任務(wù)  
        }  
    }  
  
    private static void executeTask(String task) {  
        System.out.println("Executing task: " + task);  
        // 在這里添加實(shí)際的任務(wù)執(zhí)行邏輯  
    }  
}

import org.redisson.Redisson;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;

import java.util.concurrent.TimeUnit;

public class RDelayedQueueDemo {

    public static void main(String[] args) throws InterruptedException {
        // 創(chuàng)建 Redisson 客戶端
        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        RedissonClient redisson = Redisson.create(config);

        // 獲取延遲隊(duì)列
        RDelayedQueue<String> delayedQueue = 
            redisson.getDelayedQueue("delayedQueue");

        // 添加延遲任務(wù)
        delayedQueue.offer("task1", 5, TimeUnit.SECONDS);

        // 監(jiān)聽并處理延遲任務(wù)
        Thread listenerThread = new Thread(() -> {
            while (true) {
                try {
                    // 通過 take 方法等待并獲取到期的任務(wù)
                    String task = delayedQueue.take();
                    System.out.println("Handle task: " + task);
                } catch (InterruptedException e) {
                    break;
                }
            }
        });
        listenerThread.start();
    }
}