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

测试代码：

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