redis消息队列如何延时 redis实现延时消息队列

文章摘要：redis消息队列如何延时 redis实现延时消息队列

redis消息队列延时的示例： 延时队列可通过zset来实现，消息的处理时间作为score，最后通过多线程轮询 […]

redis消息队列延时的示例：

延时队列可通过zset来实现，消息的处理时间作为score，最后通过多线程轮询获取到期的score任务即可，代码：

public class DelayQueue {

static class TaskItem {

public String id;

public T msg;

}

private Type taskType = new TypeReference>() {

}.getType();

private Jedis jedis;

private String queueName;

public DelayQueue(Jedis jedis, String queueName) {

this.jedis = jedis;

this.queueName = queueName;

}

public void delay(T msg, long delayTime) {

TaskItem task = new TaskItem<>();

task.id = UUID.randomUUID().toString();

task.msg = msg;

jedis.zadd(queueName, System.currentTimeMillis() + delayTime, JSON.toJSONString(task));

}

public void loop() {

while (!Thread.interrupted()) {

Set set = jedis.zrangeByScore(queueName, 0, System.currentTimeMillis(), 0, 1);

if (set.isEmpty()) {

try {

Thread.sleep(500);

} catch (InterruptedException e) {

break;

}

continue;

}

String s = set.iterator().next();

if (jedis.zrem(queueName, s) > 0) {

TaskItem task = JSON.parseObject(s, taskType);

System.out.println(task.msg);

}

}

}

}

2.测试代码：

public static void main(String[] args) throws InterruptedException {

Jedis jedis = new Jedis("127.0.0.1", 6379);

DelayQueue delayQueue = new DelayQueue<>(jedis, "delay_queue");

Thread producer = new Thread(() -> {

for (int i = 0; i < 10; i++)

delayQueue.delay("Mr.Wang's Hub" + i, 5000);

});

Thread consumer = new Thread(() -> {

delayQueue.loop();

});

consumer.start();

producer.start();

while (Thread.activeCount() > 1)

Thread.yield();

}