延時消息隊列可以使用redis的zset來實現(xiàn)芍殖，將消息序列化為一個字符串作為zset的value左驾，消息到期時間作為zset的score确丢，然后使用多線程輪循zset獲取到期的任務進行處理霎箍，多線程是為了保證可用性，一個線程掛掉之后其他線程可以繼續(xù)處理魏保。因為有多線程熬尺，所有需要考慮并發(fā)搶任務，確保任務不會被多次執(zhí)行囱淋。
Java版本代碼示例：
public class RedisDelayingQueue {
private RedisTemplate redisTemplate;
private String queueKey; //zset鍵
public RedisDelayingQueue(RedisTemplate redisTemplate, String queueKey) {
this.redisTemplate = redisTemplate;
this.queueKey = queueKey;
}
//存數(shù)據(jù)方法
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));
}
//取數(shù)據(jù)方法
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方法可能會被多個線程調(diào)用，所以要通過remove來決定唯一的屬主
if(redisTemplate.opsForZSet().remove(queueKey,s)>0){
//搶到了 反序列化
RedisTaskItem taskItem = JSONObject.parseObject(s,RedisTaskItem.class);
System.out.println(taskItem.msg);
}
}
}
}
//redis數(shù)據(jù)實體類
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>>>>>>>>>>>>>>>>>>>>>>>>");
}
}
這種方式簡單的實現(xiàn)了一個延時消息隊列餐塘，但是同一個任務可能會被多個線程使用remove爭搶妥衣，再決定歸屬，造成了資源浪費戒傻，可以考慮使用lua scripting來進行優(yōu)化税手，將rangeByScore和remove一同挪到服務器端進行原子化操作，這樣多個進程爭搶任務是不會造成這種浪費需纳。
