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