一口气说出六种实现延时消息的方案("六种高效实现延时消息的方法全解析")

原创

ithorizon 7个月前 (10-20) 阅读数 15 #后端开发

六种高效实现延时消息的方法全解析

一、基于时间轮的延时消息实现

时间轮（Timewheel）是一种常见的数据结构，用于处理定时任务和延时消息。它通过一个固定大小的轮来组织时间，每个槽位对应一个时间间隔。

实现方法如下：


// 假设使用Java实现
public class TimeWheel {
    private final long tickMs; // 时间轮的精度，即每个槽位的时间间隔
    private final long wheelSize; // 时间轮大小
    private final long interval; // 时间轮的总体时间范围
    private final TimerTask[] buckets; // 时间轮的槽位，存储定时任务
    public TimeWheel(long tickMs, long wheelSize) {
        this.tickMs = tickMs;
        this.wheelSize = wheelSize;
        this.interval = tickMs * wheelSize;
        this.buckets = new TimerTask[(int) wheelSize];
    }
    // 添加任务到时间轮
    public void addTask(long delayMs, TimerTask task) {
        long bucketIndex = (System.currentTimeMillis() + delayMs) / tickMs % wheelSize;
        buckets[(int) bucketIndex].add(task);
    }
}

二、基于优先队列的延时消息实现

优先队列（Priority Queue）是一种按照优先级排序的队列，可以实现延时消息的有序处理。

实现方法如下：


// 假设使用Java实现
public class DelayMessageQueue {
    private PriorityQueue queue = new PriorityQueue<>(Comparator.comparingLong(m -> m.execTime));
    public void addMessage(DelayMessage message) {
        queue.offer(message);
    }
    public void processMessages() {
        while (!queue.isEmpty()) {
            DelayMessage message = queue.peek();
            if (message.execTime <= System.currentTimeMillis()) {
                queue.poll();
                // 处理消息
                processMessage(message);
            } else {
                break;
            }
        }
    }
    private void processMessage(DelayMessage message) {
        // 处理消息的逻辑
    }
}
class DelayMessage {
    long execTime;
    String message;
}

三、基于定时任务的延时消息实现

定时任务（Scheduled Task）是利用系统提供的定时任务调度器来实现的延时消息。

实现方法如下：


// 假设使用Java的ScheduledExecutorService实现
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
public void scheduleMessage(String message, long delay) {
    scheduler.schedule(() -> {
        // 处理消息
        processMessage(message);
    }, delay, TimeUnit.MILLISECONDS);
}
private void processMessage(String message) {
    // 处理消息的逻辑
}

四、基于消息队列的延时消息实现

消息队列（Message Queue）如Kafka、RabbitMQ等，可以结合消息的延迟投递功能实现延时消息。

实现方法如下：


// 假设使用RabbitMQ实现
public void sendMessage(String message, long delay) {
    // 创建一个延迟队列
    Channel channel = connection.createChannel();
    channel.exchangeDeclare("delay_exchange", "direct", true);
    channel.queueDeclare("delay_queue", true, false, false, Map.of("x-dead-letter-exchange", "", "x-message-ttl", delay));
    // 发送消息到延迟队列
    channel.basicPublish("delay_exchange", "delay_queue", new AMQP.BasicProperties.Builder().build(), message.getBytes());
}
public void consumeMessage() {
    // 消费延迟队列的消息
    Channel channel = connection.createChannel();
    channel.basicConsume("delay_queue", true, (consumerTag, message) -> {
        // 处理消息
        processMessage(new String(message.getBody()));
    }, consumerTag -> {});
}

五、基于时间戳的延时消息实现

基于时间戳的延时消息实现，是通过在消息中嵌入时间戳，然后采取时间戳进行排序处理。

实现方法如下：


// 假设使用Java实现
public class TimestampDelayMessage {
    private long timestamp;
    private String message;
    public TimestampDelayMessage(long timestamp, String message) {
        this.timestamp = timestamp;
        this.message = message;
    }
    // 获取消息的执行时间
    public long getExecTime() {
        return timestamp;
    }
    // 处理消息
    public void process() {
        // 处理消息的逻辑
    }
}
public class DelayMessageHandler {
    private PriorityQueue queue = new PriorityQueue<>(Comparator.comparingLong(TimestampDelayMessage::getExecTime));
    public void addMessage(TimestampDelayMessage message) {
        queue.offer(message);
    }
    public void processMessages() {
        while (!queue.isEmpty()) {
            TimestampDelayMessage message = queue.peek();
            if (message.getExecTime() <= System.currentTimeMillis()) {
                queue.poll();
                message.process();
            } else {
                break;
            }
        }
    }
}

六、基于Redis的延时消息实现

Redis的Sorted Set结构可以用来实现延时消息，通过设置Sorted Set的score为消息的执行时间戳。

实现方法如下：


// 假设使用Java的Jedis实现
public void addMessage(String message, long delay) {
    long execTime = System.currentTimeMillis() + delay;
    jedis.zadd("delay_messages", execTime, message);
}
public void processMessages() {
    long currentTime = System.currentTimeMillis();
    Set messages = jedis.zrangeByScore("delay_messages", 0, currentTime);
    for (String message : messages) {
        // 移除已处理的消息
        jedis.zrem("delay_messages", message);
        // 处理消息
        processMessage(message);
    }
}
private void processMessage(String message) {
    // 处理消息的逻辑
}