延时消息队列可以使用redis的zset来实现，将消息序列化为一个字符串作为zset的value，消息到期时间作为zset的score，然后使用多线程轮循zset获取到期的任务进行处理，多线程是为了保证可用性，一个线程挂掉之后其他线程可以继续处理。因为有多线程，所有需要考虑并发抢任务，确保任务不会被多次执行。
Java版本代码示例：
public class RedisDelayingQueue {
private RedisTemplate redisTemplate;
private String queueKey; //zset键
public RedisDelayingQueue(RedisTemplate redisTemplate, String queueKey) {
this.redisTemplate = redisTemplate;
this.queueKey = queueKey;
}
//存数据方法
public void delay(String msg){
RedisTaskItem item = new RedisTaskItem();
item.id = UUIDUtil.getUUID();
item.msg = msg;
String s = JSONObject.toJSONString(item);
redisTemplate.opsForZSet().add(queueKey,s,Double.valueOf(System.currentTimeMillis() + 5000));
}
//取数据方法
public void loop(){
while (!Thread.interrupted()){
//每次只取一条
Set<Object> values = redisTemplate.opsForZSet().rangeByScore(queueKey,0L,System.currentTimeMillis(),0L,1L);
if(values.isEmpty()){
try {
Thread.sleep(500);
} catch (InterruptedException e) {
break;
}
continue;
}
String s = values.iterator().next().toString();
//loop方法可能会被多个线程调用，所以要通过remove来决定唯一的属主
if(redisTemplate.opsForZSet().remove(queueKey,s)>0){
//抢到了 反序列化
RedisTaskItem taskItem = JSONObject.parseObject(s,RedisTaskItem.class);
System.out.println(taskItem.msg);
}
}
}
}
//redis数据实体类
public class RedisTaskItem {
public String id;
public String msg;
}
测试方法
@Resource(name = "redisTemplate0")
private RedisTemplate<String, Object> redisTemplate;//使用redisTemplate操作redis
@Test
public void redisTest(){
RedisDelayingQueue queue = new RedisDelayingQueue(redisTemplate,"q-demo");
Thread product = new Thread(){
@Override
public void run() {
for (int i = 0; i < 10; i++) {
queue.delay("choose" + i);
}
}
};
Thread consumer = new Thread(){
@Override
public void run() {
queue.loop();
}
};
System.out.println("开始时间为" + new Date());
product.start();
consumer.start();
try {
product.join();
Thread.sleep(5100);
consumer.interrupt();
consumer.join();
System.out.println("结束时间为" + new Date());
}catch (InterruptedException e){
System.out.println("error>>>>>>>>>>>>>>>>>>>>>>>>");
}
}
这种方式简单的实现了一个延时消息队列，但是同一个任务可能会被多个线程使用remove争抢，再决定归属，造成了资源浪费，可以考虑使用lua scripting来进行优化，将rangeByScore和remove一同挪到服务器端进行原子化操作，这样多个进程争抢任务是不会造成这种浪费。
